我在8月底的时候，发过一篇基于 LlamaIndex+LangChain 框架，开发的简历筛选助手的应用。后续有星球成员提出希望能增加多个岗位的管理功能，正好接下来的校招活动可以用的上。

这篇在原项目的基础上，核心实现了多岗位并行管理（独立 JD、候选人池、向量索引隔离）和 HR 工作流（标签系统、分组展示、快速操作），同时进行了架构重构（分层设计、数据分库、模块化），并增强了大模型分析输出（四级推荐等级、结构化优劣势）和智能问答（按岗位过滤检索、流式输出）。

这篇试图说清楚：

系统的实际效果演示、四层系统架构拆解、五点核心技术实现、三个二次开发场景指南，以及对端侧模型应用的一些感想。

1、视频效果展示

视频查阅下方原链接：

https://mp.weixin.qq.com/s/0CuoiCagmG25rKjB-Rf8fw

2、系统架构设计

在开始讲具体实现之前，老规矩先来看看整个系统的架构设计。下面这张图展示了从用户界面到数据存储的完整数据流。

2.1、为啥要分四层

在上一版单岗位系统里，UI 代码、业务逻辑、AI 调用全都混在一起的，一开始写起来确实快，但这次升级到多岗位管理的时候，改动起来难免顾此失彼，所以这次也算是做了下系统重构。

前端交互层这部分依然采用了轻量化的 Streamlit，搭建了三个 Tab 页面：候选人概览、候选人详情和智能问答。这一层只管显示数据和响应用户操作，不关心数据从哪来以及怎么处理。

业务服务层这部分是整个系统的核心处理逻辑所在。一份简历从上传到展示，需要经过：先提取文本，再调用大模型分析，然后存入数据库，最后建立向量索引。这些流程编排都在ResumeProcessor这个类里完成。还有一个PositionService，专门管理岗位的增删改查。这一层的好处是，如果以后想改业务流程，比如在大模型分析前加一个简历去重检查，只需要在这一层加几行代码，不用动其他地方。

核心引擎层这部分封装了所有大模型相关的能力。这一层最重要的是RAGEngine，包括了简历文本向量化、存入 ChromaDB，以及在用户提问时检索相关内容等功能。

数据存储层这部分用了三种存储方式：SQLite 存储岗位信息和候选人结构化数据，ChromaDB 存储向量索引，文件系统存储原始简历文件。所有复杂的 SQL 逻辑都封装在各个 Store 类里，这样以后如果要把 SQLite 换成 MySQL，只需要改 Store 类的实现，上层代码完全不用动。

2.2、多岗位数据隔离

为了保证不同岗位的数据不会串台，我在三个层面做了隔离设计。

文件系统按岗位分目录

最简单直接的办法，就是把不同岗位的简历文件分开存。我在uploaded_resumes/目录下，给每个岗位创建一个子目录，目录名就是岗位 ID。比如"AI 产品经理"的岗位 ID 是position_001，这样做的好处是删除岗位的时候可以直接把整个目录删掉。

关系数据库用外键关联

在 SQLite 里，我给candidates表、ai_analysis表、hr_tags表都加上了position_id字段，并且设置了外键约束。每次查询候选人数据，SQL 语句里必须带上WHERE position_id = ?，从源头上避免跨岗位查询。

向量数据库按岗位分 collection

这是整个隔离机制里最关键的一环，ChromaDB 支持创建多个 collection（可以理解为不同的向量数据库），我给每个岗位创建一个独立的 collection。比如position_001对应的 collection 叫resumes_position_001,position_002对应的叫resumes_position_002。

可能会有人问为啥不用 metadata 过滤呢？比如所有简历存在一个 collection 里，然后在 metadata 里标记position_id，检索的时候再过滤。这个方案看起来更简单，但有个致命问题是，实际使用的时候性能会随着候选人总数线性下降。假设系统里有 10 个岗位，每个岗位 100 个候选人，那全局就有 1000 个候选人的向量。每次检索都要在 1000 个向量里搜索，然后再用 metadata 过滤，这无疑会很慢。

而用 collection 隔离，每个岗位的检索只在自己那 100 个向量里进行，性能只跟该岗位的候选人数相关，跟其他岗位完全无关。这就是物理隔离优于逻辑过滤的典型场景。这点非常像我在做 ibm rag 冠军赛项目拆解中提到的“疑一文一库”的做法。感兴趣的可以对照下看看。

详见同名公众号文章：

《IBM RAG挑战赛冠军方案第二弹：像素级复现→7大工程经验》

https://mp.weixin.qq.com/s/OgyrKl8qiEuZIVUvyip6eQ

2.3、RAG 引擎与业务逻辑的解耦

在上一版系统里，我把 RAG 的代码直接写在业务逻辑里，结果就是代码复用性很差。这次重构我专门抽出了一个通用的RAGEngine类，只提供三个核心能力:建索引、检索、问答。

业务层想用的时候，把数据准备好，调用对应的方法就行。比如ResumeProcessor在处理简历的时候，会调用RAGEngine.build_index()把简历文本向量化；在智能问答的时候，会调用RAGEngine.query_with_rag()执行 RAG 流程。

这种解耦带来的好处是，如果各位想换一个向量数据库，比如从 ChromaDB 换成 Milvus，只需要改RAGEngine的内部实现，业务层的代码一行都不用动。或者如果想把这套 RAG 引擎用到其他项目，比如做一个标准的企业知识库问答系统，直接复制core/rag_engine_v2.py这个文件过去，写一个新的 Processor 类就能跑起来。

3、核心技术实现拆解

前面讲完了架构设计，这部分从代码层面讲几个关键的技术实现，这部分会聚焦在私以为有一定工程借鉴价值的地方。

3.1、简历完整向量化的考量

一个好用的 RAG 系统设计，一个绕不开问题是如何精准的切分文档。上一版系统里，因为目的是要演示完整的系统流程，所以演示的简历部分也是针对性的进行了设计，分块部分按照"核心技能"、"工作经历"等章节标题进行处理。但这显然不符合实际五花八门的简历格式情况。 这次重构，我做了一个看似偷懒实则更务实的做法，就是把每份简历作为一个完整的 node，不做分块处理。

class MultiPositionNodeParser(NodeParser):
    """
    支持多岗位的简历Node解析器(无分块版本)

    设计策略:
    - 每份简历作为一个完整的node,不进行分块
    - 在metadata中添加position_id用于多岗位隔离
    - 添加candidate_name用于候选人识别

    优势:
    - 保留完整上下文,避免信息碎片化
    - 简历通常很短,适合整体向量化
    - 检索时一次性获得候选人所有信息
    """

    def _parse_nodes(self, documents: List[Document], **kwargs) -> List[BaseNode]:
        all_nodes = []
        position_id = kwargs.get('position_id')

        # 按文件路径分组(一个PDF可能有多页)
        docs_by_filepath = defaultdict(list)
        for doc in documents:
            docs_by_filepath[doc.metadata.get("file_path")].append(doc)

        for file_path, doc_parts in docs_by_filepath.items():
            # 合并所有页面为完整文本
            doc_parts.sort(key=lambda d: int(d.metadata.get("page_label", "0")))
            full_text = "\n\n".join([d.get_content().strip() for d in doc_parts])

            # 创建包含整份简历的node
            metadata = {
                "position_id": position_id,
                "candidate_name": Path(file_path).stem,
                "chunk_type": "full_resume",
                "resume_length": len(full_text)
            }

            node = TextNode(text=full_text, metadata=metadata)
            all_nodes.append(node)

        return all_nodes

首先有个共识是，简历通常很短，一般 2-4 页，毛估 2000-4000 字左右，完全在主流嵌入模型的处理范围内。这次演示用的bge-m3模型支持 8192 tokens，处理一份完整简历应该说毫无压力。

其次，完整向量化也避免了信息碎片化。当 HR 问“张三有没有 RAG 项目经验”的时候，如果简历被切成 10 个 chunk，可能需要检索多个 chunk 才能拼凑出完整答案。而整份简历作为一个 node，一次检索就能拿到所有相关信息，LLM 可以看到完整上下文进行推理。

再者，实现逻辑简单也减少了 bug 风险。代码的核心逻辑就是把多页 PDF 合并成一个字符串，然后塞进一个 node，没有复杂的正则匹配、没有边界判断，代码清晰好维护。在实际使用中实测效果很好，检索准确率明显提升，而且及时是 8b 尺寸的量化模型推理过程也很稳定。

3.2、Pydantic 驱动的结构化分析

如何让 LLM 稳定地输出结构化数据，这也是个绕不开的挑战。好的工程实践，无非也是一套围绕模型动态边界构建的解决方案。这次我用了在历史企业项目中常用的 Pydantic 模型 + 详细 Prompt 的组合，实现了相对可靠的结构化输出。先定义数据模型:

class AIResumeAnalysis(BaseModel):
    """AI简历分析结果 - 定性分析 + 结构化提取"""

    recommendation_level: str = Field(
        default="可考虑",
        description="推荐等级: 强烈推荐 | 推荐 | 可考虑 | 不推荐"
    )

    key_strengths: List[str] = Field(
        default_factory=list,
        description="具体、可验证的优势,优先列出与岗位强相关的亮点"
    )

    key_concerns: List[str] = Field(
        default_factory=list,
        description="需要关注的方面,诚实指出但不夸大"
    )

    one_sentence_summary: str = Field(
        default="",
        description="核心特征概括,帮助HR快速建立印象"
    )

    total_years_experience: int = Field(default=0)
    work_experience: List[WorkExperienceExtracted] = Field(default_factory=list)
    project_experience: List[ProjectExperienceExtracted] = Field(default_factory=list)

    # 字段校验器
    @field_validator('recommendation_level')
    @classmethod
    def validate_level(cls, v):
        valid_levels = ["强烈推荐", "推荐", "可考虑", "不推荐"]
        if v not in valid_levels:
            return "可考虑"
        return v

    @field_validator('key_strengths')
    @classmethod
    def validate_strengths_count(cls, v):
        if not v or len(v) == 0:
            return ["请查看简历原文进行人工评估"]
        return v[:5]  # 最多5条

Pydantic 的好处是，它不仅定义了数据结构，还内置了校验逻辑。比如recommendation_level必须是四个等级之一，key_strengths至少有 1 条最多 5 条，这些规则会自动执行。然后在 Prompt 中明确输出格式:

RESUME_ANALYSIS_PROMPT = """你是一位专业的招聘顾问,请根据岗位描述和候选人简历,输出结构化评估。# 输出要求请严格按照以下 JSON 格式输出,不要添加任何其他文字:{    "recommendation_level": "强烈推荐",    "key_strengths": [        "5年 AI 产品经验,主导过 3 个 RAG 项目成功上线",        "具备完整的 B 端产品设计能力(需求分析 → 原型设计 → 上线运营)"    ],    "key_concerns": [        "缺乏制造业行业背景(现有经验集中在互联网行业)"    ],    "one_sentence_summary": "技术型 AI 产品专家,RAG 项目经验丰富,需补足制造业行业背景",    ...}# 评估标准### key_strengths(关键优势)- 输出 3-5 条具体、可验证的优势- 优先列出与岗位强相关的亮点- 尽量包含数据支撑(如"5年经验"、"主导3个项目")- 避免空洞描述(❌"能力强" ✅"5年AI产品经验,主导3个RAG项目")..."""

Prompt 里不仅给出了 JSON 格式，还详细说明了每个字段的要求，以及提供了好例子和坏例子的对比。这种高指令性的 Prompt 实测可以明显提升 LLM 输出的质量和稳定性。最后在解析时做好容错处理:

def _parse_response(self, response_text: str) -> AIResumeAnalysis:
    """解析LLM响应为AIResumeAnalysis对象"""
    try:
        # 尝试1:直接解析
        data = json.loads(response_text)
        return AIResumeAnalysis(**data)
    except json.JSONDecodeError:
        # 尝试2:提取JSON块
        json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
        if json_match:
            data = json.loads(json_match.group(0))
            return AIResumeAnalysis(**data)

        # 尝试3:清理后解析(移除markdown代码块等)
        cleaned = self._clean_json_text(response_text)
        if cleaned:
            data = json.loads(cleaned)
            return AIResumeAnalysis(**data)

    # 降级策略:返回默认值
    return self._get_fallback_analysis()

这套机制总结来说有三层容错：先尝试直接解析，不行就用正则提取 JSON 块，再不行就清理格式后解析，最后还有一个降级策略返回默认值。实际使用中，绝大部分情况第一次就能成功解析，少数格式异常的也能被后续逻辑兜住，系统健壮性非常高。

3.3、简历处理的五步流水线

一份简历从上传到最终可用，需要经历多个步骤。原版系统这些逻辑散落在各处，这次我用ResumeProcessor类把整个流程标准化成五步流水线：

def process_uploaded_file(self, uploaded_file, position_id, job_description):
    """处理上传的简历文件(完整流程)"""

    result = {
        'steps': {
            'extract': False,   # 步骤1:文本提取
            'validate': False,  # 步骤2:内容验证
            'analyze': False,   # 步骤3:AI分析
            'index': False,     # 步骤4:向量索引
            'save': False       # 步骤5:保存数据
        }
    }

    # 步骤1:文本提取
    text, extract_success = extract_text_from_file(temp_path)
    result['steps']['extract'] = extract_success
    if not extract_success:
        return False, f"❌ 文件解析失败", result

    # 步骤2:内容验证
    is_valid = validate_resume_content(text)
    result['steps']['validate'] = is_valid
    if not is_valid:
        return False, f"⚠️ 文件内容疑似不是简历", result

    # 步骤3:AI分析
    analysis, analyze_success = self.analyzer.analyze(text, job_description)
    result['steps']['analyze'] = analyze_success

    # 步骤4:向量索引
    index_success = self.rag_engine.ingest_resume(temp_path, position_id)
    result['steps']['index'] = index_success

    # 步骤5:保存到数据库
    profile = self._create_candidate_profile(...)
    candidate_id = self.candidate_store.save(profile, analysis)
    result['steps']['save'] = candidate_id > 0

    return True, "✅ 处理成功", result

这个设计有几个好处：

首先，每一步都有明确的输入输出和成功标志。result['steps']字典记录了每一步的执行状态,方便调试和监控。如果处理失败,可以立刻看到是在哪一步出的问题。

其次，失败快速返回,避免无效计算。如果文本提取就失败了，后面的大模型分析、向量索引都不用做了，直接返回错误信息。这种 fail-fast 策略节省资源,也让错误信息更清晰。

最后，result字典不仅包含每一步的状态，还包含候选人 ID、分析结果、文件路径等信息，上层 UI 可以根据这些信息给用户精准的反馈。

这种流水线模式在企业级应用中非常常见，它把复杂流程拆解成清晰的步骤，每步职责单一，容易测试和维护。

3.4、多岗位检索的元数据过滤机制

前面架构部分提到，我用 metadata 过滤实现多岗位隔离。这里展示一下具体的检索代码:

def retrieve(self, query: str, position_id: int = None, top_k: int = 5):
    """检索相关文档节点"""

    # 构建metadata过滤器
    filters = None
    if position_id is not None:
        filters = MetadataFilters(
            filters=[
                MetadataFilter(
                    key="position_id",
                    value=position_id,
                    operator=FilterOperator.EQ
                )
            ]
        )

    # 创建检索器
    retriever = self.index.as_retriever(
        similarity_top_k=top_k,
        filters=filters
    )

    # 执行检索
    retrieved_nodes = retriever.retrieve(query)
    return retrieved_nodes

LlamaIndex 的MetadataFilters功能非常强，支持多种操作符(EQ、GT、LT、IN等)，可以组合多个条件。这里我只用了最简单的等值匹配，但如果以后需要更复杂的查询，比如"工作年限大于 5 年且有制造业背景"，只需要添加更多的MetadataFilter就行。在向量化这一步，我把position_id、candidate_name等信息存入 node 的 metadata:

metadata = {
    "position_id": position_id,
    "candidate_name": Path(file_path).stem,
    "chunk_type": "full_resume",
    "resume_length": len(full_text)
}
node = TextNode(text=full_text, metadata=metadata)

检索的时候，ChromaDB 会先在全局向量空间找到语义最相似的 top-K 个结果，然后用 metadata 过滤器筛选出符合条件的 node。语义匹配 + 精确过滤的组合，既保证了检索质量,又实现了数据隔离。

需要注意的是，我在架构设计部分提到过 collection 隔离性能更好。但实际开发时我发现 LlamaIndex 对多 collection 管理不太友好，需要为每个岗位创建独立的 index 对象，代码会变得很复杂。所以我在单 collection + metadata 过滤和多 collection 隔离之间做了权衡,选择了前者。

这也是说明了理论最优方案不一定是工程最优方案。要考虑框架限制、开发成本、可维护性等多方面因素。目前这个方案在候选人数量不超过 200 的情况下性能完全够用，如果以后数据量真的上去了，再重构成多 collection 也不迟。

3.5、基于 Session 的多轮对话记忆

智能问答如果只能单轮回答，用户体验无疑会很差。我用 Streamlit 的session_state实现了按岗位隔离的对话记忆。

def render_chat_tab(position_id: int, position_name: str, ...):
    """渲染智能问答Tab"""

    # 初始化聊天历史 - 每个岗位独立的对话记忆
    chat_key = f"chat_history_{position_id}"
    if chat_key not in st.session_state:
        st.session_state[chat_key] = []

    # 显示历史消息
    recent_messages = st.session_state[chat_key][-15:]  # 只显示最近15条
    for message in recent_messages:
        with st.chat_message(message["role"]):
            st.markdown(message["content"])

    # 用户输入
    if prompt := st.chat_input("请在此输入您的问题..."):
        # 添加用户消息到历史
        st.session_state[chat_key].append({"role": "user", "content": prompt})

        # ... 执行RAG检索和LLM推理 ...

        # 保存AI回答到历史
        st.session_state[chat_key].append({
            "role": "assistant",
            "content": answer_content,
            "think_content": think_content  # 推理过程
        })

首先是按岗位 ID 隔离对话历史。不同岗位的问答互不干扰，HR 在"AI 产品经理"岗位的对话，不会影响到"算法工程师"岗位。其次，只显示最近 N 条消息。对话历史完整保存在session_state里，但页面上只显示最近 15 条，避免页面过长影响体验。

最后，保存推理过程。除了最终答案，我还把 LLM 的 thinking 过程保存下来。这样用户可以在st.expander里查看 AI 的推理链路,提升可解释性。这个实现虽然简单，但在实际使用中效果很好。多轮对话让 HR 可以不断追问，获得更深入的候选人情况了解。

4、二次开发指南

在实际使用中，不同企业或者用户会有不同的定制需求。这部分我根据和一些从业者的沟通，介绍三个高频需求场景，作为二次开发的参考。

4.1、批量导入简历

从招聘网站批量下载的简历通常打包在 zip 文件里，每次都要手动解压再上传，非常低效。如果能直接上传 zip 包，系统自动解压并批量处理，能节省掉不必要的手动操作。

# 核心逻辑:解压zip并批量处理
import zipfile
from pathlib import Path
if uploaded_file.name.endswith('.zip'):
    # 解压到临时目录
    with zipfile.ZipFile(uploaded_file) as zip_ref:
        zip_ref.extractall('./temp_extract')

    # 扫描所有简历文件
    resume_files = list(Path('./temp_extract').rglob('*.pdf')) + \
                   list(Path('./temp_extract').rglob('*.docx'))

    # 调用现有的批量处理
    processor.batch_process(resume_files, position_id, job_description)

核心实现思路是在上传组件支持 zip 格式，解压后递归扫描所有简历文件，然后调用现有的批量处理逻辑即可。

4.2、薪资期望范围提取与预算匹配

薪资是招聘决策中的关键因素，但人工查看每份简历判断是否在预算内，无疑会浪费大量时间在预算不匹配的候选人上。如果系统能自动提取薪资期望并与岗位预算对比，就可以在筛选阶段就过滤掉不合适的候选人。

# 1. 扩展数据模型
class AIResumeAnalysis(BaseModel):
    # ... 原有字段 ...
    expected_salary_min: int = Field(default=0, description="期望月薪下限(K)")
    expected_salary_max: int = Field(default=0, description="期望月薪上限(K)")
# 2. 在Prompt中增加提取指令(analysis_prompts.py)
# "从简历中提取薪资期望,统一转为月薪,如'20-25K'或'年薪30万'→25K"
# 3. 在UI中显示匹配状态
if candidate.expected_salary_max <= position.salary_budget_max:
    st.success(f"✅ 预算内 ({candidate.expected_salary_min}-{candidate.expected_salary_max}K)")

核心实现思路，是在大模型分析的 Pydantic 模型中增加薪资字段，在 Prompt 中增加提取指令，大模型会自动识别各种薪资表述("20-25K"、"年薪 30 万"等)并归一化为统一格式。

4.3、AI 生成面试问题清单

筛选出候选人后需要准备面试问题，要仔细读简历找出可以深挖的点。如果系统能根据简历和岗位要求自动生成针对性的面试问题，可以大幅提升面试准备效率。

# 生成面试问题的核心Prompt
PROMPT = """基于简历和岗位要求,生成面试问题:
- 能力验证: 验证简历中的技能是否真实
- 项目深挖: 了解关键项目的实际贡献
- 短板确认: 针对"{key_concerns}"提问
输出JSON: {{"ability": [...], "project": [...], "weakness": [...]}}
"""
# 调用LLM生成
response = llm.complete(PROMPT.format(key_concerns=candidate.ai_concerns))
questions = json.loads(response.text)
# 在详情页展示
st.subheader("💬 AI生成的面试问题")
for category, qs in questions.items():
    for q in qs:
        st.write(f"• {q}")

核心实现思路是设计一个专门的 Prompt，让 LLM 基于简历内容和分析结果，生成分类的面试问题(能力验证、项目深挖、短板确认等)。

5、写在最后

现在，当大家谈论大模型企业应用落地的时候，默认的潜台词都是企业主导、集中部署。而现实情况是，企业去落地一个面向于不同部门的大模型应用，是一个道阻且长的过程。但实际上像 HR、律师、会计这类专业工作者，每天也在做大量重复劳动。如果能把这套系统打包成一个开箱即用的桌面应用，内置嵌入模型、向量数据库、开源 LLM，完全本地运行，不依赖云服务，可以更加短平快的的给很多岗位带来提效或者解放双手。

这个项目中演示时使用的 qwen3:8b(Q4 量化版)5.2GB大小，在我的 24GB 内存 MacBook 上，首字响应时间大概 5 秒，后续 token 生成速度相对比较流畅。但这个性能对大部分 HR 的工作电脑来说是个不小的挑战。DeepSeek 年初开源的蒸馏后的小尺寸模型，在智力水平和电脑性能要求上都不够实用，但现在似乎重新来到了端侧模型应用重新爆发的临界点。一方面，1.5B-3B 的小模型配合垂直领域微调，完全可以胜任简历筛选这类专业场景。其次，更优秀的小尺寸开源模型更新的速度我想也会超出大家的预期。

从创业视角来看，这不仅是 2B 市场的机会，更是面向专业个人用户(像 Cursor 面向开发者那样)的增量市场。让专业工作者都能拥有自己的大模型应用助手，而不是一味的等待企业采购，这或许是接下来非常值得保持关注的长尾需求。

下篇介绍ragflow集成mineru2.5集成教程，感兴趣的蹲一蹲。

本篇涉及的完整项目脚本和测试文档已上传至知识星球

https://2l1vp.xetlk.com/s/3uAmEv