【人工智能招聘】FastAPI+LangChain打造智能招聘系统-97java

┌─────────────────────────────────────────────────────────────────────┐
│                         前端（React SPA）                          │
│              HR操作台 │ 候选人门户 │ AI面试间 │ 数据分析看板        │
└─────────────────────────────┬───────────────────────────────────────┘
                              │ HTTPS + WebSocket
┌─────────────────────────────▼───────────────────────────────────────┐
│                      FastAPI 应用层（CVM集群）                       │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐    │
│  │ 简历处理模块 │  │ 匹配引擎   │  │ AI面试Agent（LangGraph） │    │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘    │
└───┬───────────────┬───────────────┬───────────────────────────────┘
    │               │               │
┌───▼───────┐ ┌────▼────┐ ┌───────▼──────┐ ┌──────────────────────┐
│  腾讯云COS │ │腾讯云OCR│ │ 腾讯云VectorDB│ │  腾讯云混元大模型API  │
│ (简历存储) │ │(简历解析)│ │ (向量检索)    │ │  (LLM推理)           │
└───────────┘ └─────────┘ └───────────────┘ └──────────────────────┘
                              │
                    ┌─────────▼─────────┐
                    │  PostgreSQL（腾讯云）│
                    │  (业务数据/用户/面试) │
                    └────────────────────┘

# resume_parser/ocr.py
import json
import base64
from tencentcloud.common import credential
from tencentcloud.ocr.v20181119 import ocr_client, models
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from loguru import logger

class TencentOCRService:
    """腾讯云OCR服务封装"""
    
    def __init__(self, secret_id: str, secret_key: str, region: str = "ap-guangzhou"):
        cred = credential.Credential(secret_id, secret_key)
        http_profile = HttpProfile()
        http_profile.endpoint = "ocr.tencentcloudapi.com"
        client_profile = ClientProfile()
        client_profile.httpProfile = http_profile
        self.client = ocr_client.OcrClient(cred, region, client_profile)
    
    def parse_resume_image(self, image_base64: str) -> dict:
        """解析简历图片（支持JPG/PNG）"""
        req = models.GeneralBasicOCRRequest()
        req.ImageBase64 = image_base64
        
        try:
            resp = self.client.GeneralBasicOCR(req)
            # 提取文本块
            text_blocks = []
            for item in resp.TextDetections:
                text_blocks.append({
                    "text": item.DetectedText,
                    "confidence": item.Confidence,
                    "polygon": item.Polygon
                })
            
            # 合并文本
            full_text = "\n".join([b["text"] for b in text_blocks])
            
            return {
                "raw_text": full_text,
                "blocks": text_blocks,
                "confidence": resp.Angle if hasattr(resp, "Angle") else 0
            }
        except Exception as e:
            logger.error(f"OCR解析失败: {e}")
            raise
    
    def parse_resume_pdf(self, cos_url: str) -> dict:
        """解析PDF简历（使用腾讯云PDF OCR）"""
        # 使用PDF OCR专用接口
        req = models.RecognizeTableOCRRequest()
        req.ImageUrl = cos_url
        # 实际代码类似，省略详细实现
        pass

# resume_parser/extractor.py
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional
from langchain_community.chat_models import ChatHunyuan

class CandidateInfo(BaseModel):
    """候选人信息结构化模型"""
    name: str = Field(description="候选人姓名")
    phone: str = Field(description="手机号码")
    email: str = Field(description="电子邮箱")
    age: Optional[int] = Field(default=None, description="年龄")
    education: List[dict] = Field(description="教育经历，包含school、major、degree、start_date、end_date")
    work_experience: List[dict] = Field(description="工作经历，包含company、position、start_date、end_date、responsibilities")
    skills: List[str] = Field(description="技能列表")
    certifications: List[str] = Field(description="证书/资质")
    summary: str = Field(description="候选人简介摘要")

class ResumeExtractor:
    def __init__(self, api_key: str = None):
        # 初始化腾讯混元
        self.llm = ChatHunyuan(
            model="hunyuan-pro",
            hunyuan_api_key=api_key or os.getenv("HUNYUAN_API_KEY"),
            temperature=0.1,  # 低温度保证稳定性
        )
        self.parser = PydanticOutputParser(pydantic_object=CandidateInfo)
        
    def extract(self, raw_text: str) -> CandidateInfo:
        """从OCR原始文本中提取结构化信息"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一名专业的简历解析专家。请从以下简历文本中提取候选人信息。
            要求：
            1. 严格按JSON格式输出
            2. 如果没有提取到信息，字段留空或null
            3. 日期统一为YYYY-MM-DD格式
            4. 技能提取要全面，包括硬技能和软技能
            
            {format_instructions}
            """),
            ("human", "简历文本：\n{resume_text}")
        ])
        
        chain = prompt | self.llm | self.parser
        
        result = chain.invoke({
            "resume_text": raw_text,
            "format_instructions": self.parser.get_format_instructions()
        })
        
        return result

# matching/vector_store.py
import tcvectordb
from tcvectordb.model.enum import FieldType, IndexType, MetricType
from tcvectordb.model.index import Index, VectorIndex, FilterIndex
from tcvectordb.model.document import Document, HNSWSearchParams
from typing import List, Dict
import numpy as np

class TencentVectorDB:
    """腾讯云向量数据库封装"""
    
    def __init__(self, url: str, key: str, username: str = "root"):
        self.client = tcvectordb.VectorDBClient(
            url=url,
            username=username,
            key=key,
            timeout=30
        )
        self.database_name = "recruitment_db"
        self.collection_name = "resume_embeddings"
        self._init_database()
    
    def _init_database(self):
        """初始化数据库和集合"""
        # 创建数据库
        try:
            self.client.create_database(self.database_name)
        except Exception:
            pass  # 数据库已存在
        
        db = self.client.database(self.database_name)
        
        # 创建集合（定义索引结构）
        index = Index()
        # 向量索引（使用HNSW算法）
        index.add(VectorIndex(
            "vector",
            dimension=768,  # 混元Embedding维度
            index_type=IndexType.HNSW,
            metric_type=MetricType.IP,  # 内积相似度
            params={
                "M": 16,
                "efConstruction": 200
            }
        ))
        # 标量索引（用于过滤）
        index.add(FilterIndex(
            "job_id", FieldType.String, IndexType.PRIMARY_KEY
        ))
        index.add(FilterIndex(
            "candidate_id", FieldType.String, IndexType.FILTER
        ))
        index.add(FilterIndex(
            "years_of_experience", FieldType.Uint64, IndexType.FILTER
        ))
        index.add(FilterIndex(
            "max_education", FieldType.String, IndexType.FILTER
        ))
        
        # 创建集合
        try:
            db.create_collection(
                name=self.collection_name,
                shard=1,
                replicas=2,
                description="简历向量存储",
                index=index
            )
        except Exception:
            pass
    
    def upsert_resume(self, resume_id: str, job_id: str, vector: List[float], 
                      metadata: Dict):
        """插入或更新简历向量"""
        db = self.client.database(self.database_name)
        collection = db.collection(self.collection_name)
        
        doc = Document(
            id=resume_id,
            vector=vector,
            job_id=job_id,
            candidate_id=metadata.get("candidate_id", ""),
            years_of_experience=metadata.get("years_of_experience", 0),
            max_education=metadata.get("max_education", ""),
            **metadata
        )
        
        collection.upsert([doc])
        
    def search_similar(self, query_vector: List[float], job_id: str, 
                       top_k: int = 10, filter_expr: str = None) -> List[Dict]:
        """相似度检索"""
        db = self.client.database(self.database_name)
        collection = db.collection(self.collection_name)
        
        # 构造过滤条件
        filter_condition = f'job_id="{job_id}"'
        if filter_expr:
            filter_condition += f" and {filter_expr}"
        
        results = collection.search(
            vectors=[query_vector],
            filter=filter_condition,
            params=HNSWSearchParams(ef=200),
            limit=top_k,
            retrieve_vector=False,
            output_fields=["candidate_id", "years_of_experience", "max_education", "skills"]
        )
        
        # 解析结果
        matches = []
        for result in results:
            for doc in result:
                matches.append({
                    "candidate_id": doc.get("candidate_id"),
                    "score": doc.get("score"),
                    "metadata": doc
                })
        
        return matches

# matching/scorer.py
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
import json

class MatchScorer:
    """人岗匹配评分器"""
    
    def __init__(self, llm):
        self.llm = llm
        
    def compute_match_score(self, jd_text: str, resume_text: str) -> dict:
        """计算JD与简历的匹配度（混元精排）"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一位资深招聘专家。请评估候选人与岗位的匹配程度。
            请从以下维度打分（1-10分）：
            1. 硬技能匹配度（技术栈、工具）
            2. 工作经验匹配度（行业、年限）
            3. 教育背景匹配度
            4. 软技能匹配度
            5. 综合匹配度
            
            输出JSON格式，包含各维度得分和详细说明。
            """),
            ("human", """
            岗位描述（JD）：
            {jd_text}
            
            候选人简历：
            {resume_text}
            
            请给出匹配度评估。
            """)
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result_text = chain.invoke({
            "jd_text": jd_text,
            "resume_text": resume_text
        })
        
        # 解析JSON
        try:
            score_data = json.loads(result_text)
        except json.JSONDecodeError:
            # 降级处理
            score_data = {
                "hard_skills": 5,
                "experience": 5,
                "education": 5,
                "soft_skills": 5,
                "overall": 5,
                "reasoning": result_text
            }
        
        return score_data

# interview/state.py
from typing import TypedDict, List, Literal, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import add_messages
from datetime import datetime

class InterviewState(TypedDict):
    """面试Agent状态"""
    messages: Annotated[List[BaseMessage], add_messages]
    candidate_id: str
    job_id: str
    current_question_index: int
    questions: List[dict]  # [{"question": "xxx", "dimension": "xxx"}]
    answers: List[dict]    # [{"question": "xxx", "answer": "xxx", "score": 0}]
    evaluation: dict       # 最终评价
    status: Literal["pending", "ongoing", "completed", "aborted"]
    start_time: datetime
    end_time: datetime = None

# interview/agent.py
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
import random

class Question(BaseModel):
    question: str = Field(description="面试问题")
    dimension: str = Field(description="考察维度: 技术/项目/沟通/逻辑")
    difficulty: str = Field(description="难度: 初级/中级/高级")
    
class InterviewEvaluation(BaseModel):
    overall_score: int = Field(ge=0, le=100, description="综合评分")
    strengths: List[str] = Field(description="优势项")
    weaknesses: List[str] = Field(description="待提升项")
    recommendation: str = Field(description="招聘建议: 强烈推荐/推荐/待定/不推荐")
    reason: str = Field(description="决策依据")

class AIInterviewAgent:
    def __init__(self, llm, vector_db, match_scorer):
        self.llm = llm
        self.vector_db = vector_db
        self.match_scorer = match_scorer
        self.graph = self._build_graph()
        
    def _build_graph(self):
        """构建LangGraph工作流"""
        workflow = StateGraph(InterviewState)
        
        # 添加节点
        workflow.add_node("generate_questions", self.generate_questions_node)
        workflow.add_node("ask_question", self.ask_question_node)
        workflow.add_node("analyze_answer", self.analyze_answer_node)
        workflow.add_node("evaluate", self.evaluate_node)
        workflow.add_node("finalize", self.finalize_node)
        
        # 设置入口
        workflow.set_entry_point("generate_questions")
        
        # 条件边：循环提问
        workflow.add_conditional_edges(
            "analyze_answer",
            self.should_continue,
            {
                "continue": "ask_question",
                "end": "evaluate"
            }
        )
        
        workflow.add_edge("generate_questions", "ask_question")
        workflow.add_edge("ask_question", "analyze_answer")
        workflow.add_edge("evaluate", "finalize")
        workflow.add_edge("finalize", END)
        
        return workflow.compile()
    
    async def generate_questions_node(self, state: InterviewState) -> InterviewState:
        """根据简历和JD生成面试题目"""
        # 从向量库获取简历和JD信息
        # ... 省略数据获取代码
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一位经验丰富的技术面试官。请根据候选人的简历和岗位要求，生成5道面试题。
            要求：
            1. 涵盖技术能力、项目经验、软技能三个维度
            2. 题目难度递进（从基础到深入）
            3. 结合候选人简历中的具体项目进行提问
            4. 避免重复和泛泛而谈
            """),
            ("human", """
            候选人简历摘要：{resume_summary}
            岗位要求：{jd_summary}
            
            请生成5道高质量的面试题。
            """)
        ])
        
        # 使用结构化输出
        structured_llm = self.llm.with_structured_output({"type": "array", "items": {"type": "object"}})
        questions = await structured_llm.ainvoke(prompt.format(
            resume_summary=state.get("resume_summary", ""),
            jd_summary=state.get("jd_summary", "")
        ))
        
        state["questions"] = questions
        state["current_question_index"] = 0
        return state
    
    async def ask_question_node(self, state: InterviewState) -> InterviewState:
        """向候选人提问"""
        idx = state["current_question_index"]
        if idx < len(state["questions"]):
            question = state["questions"][idx]["question"]
            state["messages"].append(
                AIMessage(content=f"面试官：{question}")
            )
        return state
    
    async def analyze_answer_node(self, state: InterviewState) -> InterviewState:
        """分析候选人回答质量"""
        # 获取最新回答
        last_answer = state["messages"][-1].content if state["messages"] else ""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一位专业面试评估专家。请对候选人的回答进行评分和分析。
            评分标准（1-10分）：
            - 准确性：回答是否正确
            - 完整性：是否全面
            - 逻辑性：思路是否清晰
            - 表达力：语言表达是否流畅
            """),
            ("human", """
            面试问题：{question}
            候选人回答：{answer}
            考察维度：{dimension}
            
            给出评分和改进建议。
            """)
        ])
        
        # 调用混元分析
        result = await self.llm.ainvoke(prompt.format(
            question=state["questions"][state["current_question_index"]]["question"],
            answer=last_answer,
            dimension=state["questions"][state["current_question_index"]]["dimension"]
        ))
        
        # 存储评分结果
        state["answers"].append({
            "question": state["questions"][state["current_question_index"]],
            "answer": last_answer,
            "analysis": result.content
        })
        
        state["current_question_index"] += 1
        return state
    
    def should_continue(self, state: InterviewState) -> str:
        """判断是否继续面试"""
        if state["current_question_index"] >= len(state["questions"]):
            return "end"
        return "continue"
    
    async def evaluate_node(self, state: InterviewState) -> InterviewState:
        """生成最终评估报告"""
        # 汇总所有问答
        qa_history = "\n".join([
            f"Q: {a['question']['question']}\nA: {a['answer']}\n分析: {a['analysis']}"
            for a in state["answers"]
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一位招聘决策专家。请根据完整面试记录，生成最终评估报告。"""),
            ("human", """
            面试完整记录：
            {qa_history}
            
            请输出：
            1. 综合评分（0-100）
            2. 候选人优势项（3-5条）
            3. 候选人待提升项（3-5条）
            4. 招聘建议（强烈推荐/推荐/待定/不推荐）
            5. 决策依据（详细说明）
            """)
        ])
        
        structured_llm = self.llm.with_structured_output(InterviewEvaluation)
        evaluation = await structured_llm.ainvoke(prompt.format(qa_history=qa_history))
        
        state["evaluation"] = evaluation.dict()
        state["status"] = "completed"
        state["end_time"] = datetime.now()
        
        return state
    
    async def finalize_node(self, state: InterviewState) -> InterviewState:
        """生成最终面试报告"""
        # 构造最终输出
        report = {
            "candidate_id": state["candidate_id"],
            "job_id": state["job_id"],
            "questions_count": len(state["questions"]),
            "answers": state["answers"],
            "evaluation": state["evaluation"],
            "duration": (state["end_time"] - state["start_time"]).total_seconds() / 60,
            "recommendation": state["evaluation"]["recommendation"]
        }
        
        # 存储到数据库
        await self.save_report(report)
        
        state["messages"].append(
            AIMessage(content=f"面试结束！综合评分：{state['evaluation']['overall_score']}分")
        )
        
        return state

# api/routes/resume.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks
from typing import List
from pydantic import BaseModel, Field

router = APIRouter(prefix="/api/v1/resume", tags=["简历管理"])

class ResumeUploadResponse(BaseModel):
    resume_id: str
    candidate_name: str
    parse_status: str
    extracted_info: dict

class ResumeMatchRequest(BaseModel):
    job_id: str
    candidate_ids: List[str] = Field(description="候选人ID列表，空则检索所有")

class MatchResponse(BaseModel):
    candidate_id: str
    match_score: float
    dimension_scores: dict
    reasoning: str

@router.post("/upload", response_model=ResumeUploadResponse)
async def upload_resume(
    file: UploadFile = File(...),
    background_tasks: BackgroundTasks = None
):
    """上传并解析简历"""
    # 1. 上传到COS
    cos_url = await upload_to_cos(file)
    
    # 2. 异步解析简历
    resume_id = generate_resume_id()
    background_tasks.add_task(parse_resume_background, resume_id, cos_url)
    
    return ResumeUploadResponse(
        resume_id=resume_id,
        candidate_name="解析中...",
        parse_status="processing",
        extracted_info={}
    )

@router.post("/match", response_model=List[MatchResponse])
async def match_resumes(request: ResumeMatchRequest):
    """简历匹配"""
    # 调用匹配引擎
    results = await match_engine.match(
        job_id=request.job_id,
        candidate_ids=request.candidate_ids
    )
    return results

@router.get("/{resume_id}/parsed")
async def get_parsed_resume(resume_id: str):
    """获取解析后的简历"""
    info = await get_candidate_info(resume_id)
    if not info:
        raise HTTPException(404, "简历未找到")
    return info

# tasks/resume_tasks.py
from celery import Celery
from loguru import logger

app = Celery("recruitment_tasks", broker=os.getenv("REDIS_URL"))

@app.task(bind=True, max_retries=3)
def parse_resume_background(self, resume_id: str, cos_url: str):
    """后台异步解析简历"""
    try:
        # 1. OCR识别
        ocr_result = ocr_service.parse_resume_pdf(cos_url)
        
        # 2. 信息提取（混元）
        extracted = resume_extractor.extract(ocr_result["raw_text"])
        
        # 3. 生成向量（混元Embedding）
        embedding = await generate_embedding(extracted.summary)
        
        # 4. 存储到向量数据库
        vector_db.upsert_resume(
            resume_id=resume_id,
            job_id=None,  # 暂未关联岗位
            vector=embedding,
            metadata=extracted.dict()
        )
        
        # 5. 更新数据库状态
        update_resume_status(resume_id, "parsed", extracted.dict())
        
        return {"resume_id": resume_id, "status": "success"}
        
    except Exception as e:
        logger.error(f"简历解析失败: {e}")
        self.retry(exc=e, countdown=60)

// components/InterviewRoom.tsx
import React, { useState, useEffect, useRef } from 'react';
import { Card, Button, Input, Avatar, Spin, Tag } from 'antd';
import { useWebSocket } from '@/hooks/useWebSocket';
import { SendOutlined, RobotOutlined, UserOutlined } from '@ant-design/icons';

interface InterviewMessage {
  role: 'interviewer' | 'candidate' | 'system';
  content: string;
  timestamp: Date;
}

export const InterviewRoom: React.FC<{ sessionId: string }> = ({ sessionId }) => {
  const [messages, setMessages] = useState<InterviewMessage[]>([]);
  const [input, setInput] = useState('');
  const [isProcessing, setIsProcessing] = useState(false);
  const messagesEndRef = useRef<HTMLDivElement>(null);
  
  // WebSocket连接
  const { sendMessage, lastMessage, readyState } = useWebSocket(
    `wss://api.example.com/ws/interview/${sessionId}`
  );
  
  useEffect(() => {
    if (lastMessage) {
      const data = JSON.parse(lastMessage);
      setMessages(prev => [...prev, {
        role: data.role,
        content: data.content,
        timestamp: new Date(data.timestamp)
      }]);
      setIsProcessing(false);
    }
  }, [lastMessage]);
  
  useEffect(() => {
    messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
  }, [messages]);
  
  const handleSend = () => {
    if (!input.trim() || isProcessing) return;
    
    // 添加用户消息
    setMessages(prev => [...prev, {
      role: 'candidate',
      content: input,
      timestamp: new Date()
    }]);
    
    // 发送到WebSocket
    sendMessage(JSON.stringify({
      type: 'answer',
      content: input,
      timestamp: new Date().toISOString()
    }));
    
    setInput('');
    setIsProcessing(true);
  };
  
  return (
    <Card className="h-full flex flex-col">
      <div className="flex items-center gap-2 mb-4">
        <RobotOutlined className="text-2xl text-blue-500" />
        <span className="font-semibold">AI面试官</span>
        <Tag color="processing">进行中</Tag>
      </div>
      
      <div className="flex-1 overflow-y-auto space-y-4 mb-4 p-4 bg-gray-50 rounded-lg">
        {messages.map((msg, idx) => (
          <div
            key={idx}
            className={`flex gap-3 ${
              msg.role === 'candidate' ? 'flex-row-reverse' : ''
            }`}
          >
            <Avatar
              icon={msg.role === 'interviewer' ? <RobotOutlined /> : <UserOutlined />}
              className={msg.role === 'interviewer' ? 'bg-blue-500' : 'bg-green-500'}
            />
            <div
              className={`max-w-[70%] p-3 rounded-lg ${
                msg.role === 'interviewer'
                  ? 'bg-white border border-gray-200'
                  : 'bg-blue-500 text-white'
              }`}
            >
              {msg.content}
              <div className="text-xs opacity-70 mt-1">
                {msg.timestamp.toLocaleTimeString()}
              </div>
            </div>
          </div>
        ))}
        
        {isProcessing && (
          <div className="flex gap-3">
            <Avatar icon={<RobotOutlined />} className="bg-blue-500" />
            <div className="bg-white border border-gray-200 p-3 rounded-lg">
              <Spin size="small" /> 思考中...
            </div>
          </div>
        )}
        <div ref={messagesEndRef} />
      </div>
      
      <div className="flex gap-2">
        <Input.TextArea
          value={input}
          onChange={(e) => setInput(e.target.value)}
          onPressEnter={(e) => {
            if (!e.shiftKey) {
              e.preventDefault();
              handleSend();
            }
          }}
          placeholder="输入你的回答... (Shift+Enter换行)"
          autoSize={{ minRows: 1, maxRows: 4 }}
          disabled={isProcessing}
          className="flex-1"
        />
        <Button
          type="primary"
          icon={<SendOutlined />}
          onClick={handleSend}
          disabled={isProcessing || !input.trim()}
        />
      </div>
    </Card>
  );
};

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: recruitment-backend
  namespace: recruitment
spec:
  replicas: 3
  selector:
    matchLabels:
      app: recruitment-backend
  template:
    metadata:
      labels:
        app: recruitment-backend
    spec:
      containers:
      - name: api
        image: ccr.ccs.tencentyun.com/recruitment/backend:latest
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
        - name: HUNYUAN_API_KEY
          valueFrom:
            secretKeyRef:
              name: hunyuan-secret
              key: api-key
        - name: VECTORDB_URL
          value: "http://vectordb-service:8080"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: recruitment-backend
  namespace: recruitment
spec:
  selector:
    app: recruitment-backend
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: recruitment-ingress
  namespace: recruitment
  annotations:
    kubernetes.io/ingress.class: "nginx"
    nginx.ingress.kubernetes.io/proxy-body-size: "50m"
    # 腾讯云CLB配置
spec:
  rules:
  - host: recruitment.example.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: recruitment-backend
            port:
              number: 80

# scf/resume_parser.py
from tencentcloud.scf import SCF
import json

def main_handler(event, context):
    """
    腾讯云函数：简历解析触发器
    """
    # 解析事件
    body = json.loads(event.get("body", "{}"))
    cos_url = body.get("cos_url")
    resume_id = body.get("resume_id")
    
    # 执行解析
    result = parse_resume(cos_url)
    
    # 回调通知
    notify_backend(resume_id, result)
    
    return {
        "statusCode": 200,
        "body": json.dumps({"resume_id": resume_id, "status": "completed"})
    }

# security/mask.py
import re

def mask_personal_info(text: str) -> str:
    """脱敏个人敏感信息"""
    # 手机号脱敏
    text = re.sub(r'(\d{3})\d{4}(\d{4})', r'\1****\2', text)
    # 身份证脱敏
    text = re.sub(r'(\d{6})\d{8}(\d{4})', r'\1********\2', text)
    # 邮箱脱敏
    text = re.sub(r'([a-zA-Z0-9]{2})[a-zA-Z0-9._-]*@([a-zA-Z0-9]+\.[a-zA-Z]+)', 
                  r'\1***@\2', text)
    return text

# security/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        role = payload.get("role")
        # 权限校验
        if role not in ["hr", "admin", "interviewer"]:
            raise HTTPException(403, "权限不足")
        return payload
    except JWTError:
        raise HTTPException(401, "无效Token")

@router.post("/interview/start")
async def start_interview(
    request: StartInterviewRequest,
    user: dict = Depends(verify_token)
):
    # 只有HR和面试官可以发起面试
    if user["role"] not in ["hr", "interviewer"]:
        raise HTTPException(403, "无权限发起面试")
    # ...

# 使用Redis缓存常见查询
from functools import lru_cache
import redis.asyncio as redis

redis_client = redis.Redis(host='redis-service', decode_responses=True)

@lru_cache(maxsize=1000)
async def get_cached_jd(job_id: str) -> str:
    """缓存JD文本（内存+Redis两级缓存）"""
    # 先查Redis
    cached = await redis_client.get(f"jd:{job_id}")
    if cached:
        return cached
    
    # 查数据库
    jd_text = await fetch_jd_from_db(job_id)
    await redis_client.setex(f"jd:{job_id}", 3600, jd_text)
    return jd_text

# 使用异步流式响应减少首字节延迟
@router.post("/interview/stream")
async def stream_interview(request: InterviewRequest):
    async def generate():
        async for chunk in interview_agent.stream(request):
            yield f"data: {json.dumps(chunk)}\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")