手把手教你搭建 AI 简历分析与 JD 解读系统（上）：架构设计与 QA 路径

本文是该系列教程的第一篇，重点讲解整体架构设计和 QA 问答路径的实现。

第二篇将覆盖子图拆分、流式输出桥接和踩坑记录。

写在前面

现在正好是金三银四中的招聘旺季，面向绝大多数求职者来说，本系统能极大提高各位求职者的简历价值以及面试能力，这是本项目创作的初衷，同时我打算将其开源，以供绝大多数求职者或者是想要学习如何构建自己的Agent的人来参考。

当然，如果想要尽快使用该系统的求职者也可以访问我的个人博客网站中的工具栏，只需注册便可体验所有功能，具体网站已放在文章末尾。

现在市面上大多数 LangGraph 教程停在一个简单的 RAG 问答 demo——用户提问，检索文档，LLM 回答，结束。

但真实业务里的 Agent 需要同时处理多种任务类型：普通问答、简历分析、岗位解读，而且每种任务有自己的节点链和数据流。如果你把所有逻辑平铺在一个图里，很快就会变成一团乱麻。

这篇文章带你从 0 搭一个智能简历编写及 JD 分析系统。我会重点讲清楚一个核心问题：

怎么用 LangGraph 把一个多路径 AI 应用编排得清晰可维护？

最终效果：一个 Web 应用，支持三种交互模式——知识库问答、简历分析评估、JD 岗位解读，每种模式有自己的处理链路，共享统一的状态管理和会话持久化。

一、系统全景

先看最终的产品结构：

┌─────────────────────────────────────────────────────┐
│                    前端 (HTML/JS)                     │
│  ┌──────────┐  ┌──────────────┐  ┌──────────────┐   │
│  │ QA 对话   │  │ 简历分析报告  │  │ JD 岗位解读  │   │
│  └──────────┘  └──────────────┘  └──────────────┘   │
└────────────────────────┬────────────────────────────┘
                         │ SSE (Server-Sent Events)
┌────────────────────────▼────────────────────────────┐
│                  FastAPI 后端                         │
│                                                      │
│  ┌────────────────────────────────────────────────┐  │
│  │              LangGraph Agent 主图                │  │
│  │                                                │  │
│  │           START                                │  │
│  │             │                                  │  │
│  │          router                                │  │
│  │        ┌───┼──────────────┐                    │  │
│  │        │   │              │                    │  │
│  │     retrieve  web      direct    resume  jd    │  │
│  │        │   │      (直答)     (子图)  (子图)     │  │
│  │     search_kb  search_web     ┌──┐   ┌──┐      │  │
│  │        │         │          │  │   │  │       │  │
│  │     normalize  normalize   │  │   │  │       │  │
│  │        │         │          └──┘   └──┘       │  │
│  │     generate  generate                          │  │
│  │        │         │                              │  │
│  │           END ←──┴──────────────────           │  │
│  └────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────┘

系统的核心是一个 LangGraph StateGraph，通过一个 router 节点做意图分类，把请求分发到不同的处理链路：

本篇聚焦前三条 QA 路径，子图部分在下一篇详解。

二、技术栈选择

在开始写代码之前，先解释为什么选这些技术：

为什么不选 Dify / Coze 等低代码平台？ 因为我们要定制化路由、多路径子图、SSE 流式输出——低代码平台做不到这种粒度的控制。

三、项目结构

先把骨架搭好：

ResumeAgent/
├── app/
│   ├── agent/                   # Agent 核心层
│   │   ├── __init__.py          # 图构建入口
│   │   ├── graph.py             # 主图定义
│   │   ├── state.py             # 状态模型
│   │   ├── prompts.py           # Prompt 模板
│   │   ├── nodes/               # 图节点
│   │   │   ├── router.py        # 意图路由
│   │   │   ├── kb_search.py     # 知识库检索
│   │   │   ├── web_search.py    # 网络搜索
│   │   │   ├── normalize.py     # 结果标准化
│   │   │   ├── generate.py      # LLM 生成
│   │   │   ├── extract_resume.py  # 简历提取（下篇讲）
│   │   │   ├── generate_analysis.py  # 简历分析生成（下篇讲）
│   │   │   └── ...
│   │   └── subgraphs/           # 子图（下篇讲）
│   │       ├── resume_analysis.py
│   │       └── jd_analysis.py
│   ├── api/
│   │   └── agent.py             # API 路由
│   ├── core/
│   │   └── config.py            # 配置管理
│   └── services/
│       ├── llm_service.py       # LLM 调用封装
│       └── retrieval_service.py # 向量检索服务
├── knowledge-base/              # 知识库文档 (Markdown)
├── static/                      # 前端文件
├── requirements.txt
└── main.py                      # 启动入口

这个结构的关键设计理念是职责分离：

• nodes/ 下的每个文件 = 一个图节点 = 一个函数
• graph.py 只负责组装节点和边，不包含业务逻辑
• state.py 定义全图共享的状态模型
• prompts.py 集中管理所有 Prompt 模板

四、Step 1：定义状态模型

State 是 LangGraph 的灵魂。所有节点通过读写同一个 State 字典来传递数据。

打开 app/agent/state.py：

from __future__ import annotations
from enum import Enum
from typing import Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
from typing_extensions import TypedDict


class RouteType(str, Enum):
    """数据来源路由"""
    RETRIEVE = "retrieve"   # 知识库检索
    WEB = "web"             # 网络搜索
    DIRECT = "direct"       # 直接回答


class TaskType(str, Enum):
    """任务类型路由"""
    QA = "qa"
    RESUME_ANALYSIS = "resume_analysis"
    JD_ANALYSIS = "jd_analysis"


class RouteDecision(BaseModel):
    """Router 结构化输出模型"""
    reasoning: str = Field(description="简要判断理由")
    route_type: RouteType = Field(description="数据来源路径")
    task_type: TaskType = Field(description="任务类型")


class AgentState(TypedDict, total=False):
    """Agent 全局状态"""

    # 消息历史（用 add_messages reducer 自动管理追加和去重）
    messages: Annotated[list[BaseMessage], add_messages]

    # 结构化检索结果（各检索节点写入，generate 节点消费）
    context_sources: list[dict]

    # 文本化上下文（normalize 节点从 context_sources 拼装后写入）
    working_context: 

    
    route_type:     
    task_type:      

    
    session_id: 

    
    resume_data:  | 

    
    jd_data:  | 

    
    final_answer: 

这里有几个关键设计决策：

1. 为什么用 TypedDict 而不是 Pydantic BaseModel？

LangGraph 的 State 需要支持 partial update——每个节点只返回它修改的字段，框架自动合并。TypedDict + total=False 天然支持这个语义。Pydantic BaseModel 会要求所有字段都有默认值，不够灵活。

2. add_messages reducer 是什么？

messages: Annotated[list[BaseMessage], add_messages]

这行代码告诉 LangGraph：当多个节点都往 messages 里写数据时，不要覆盖，而是追加。而且 add_messages 内置了去重逻辑——如果新消息的 id 和已有消息相同，会替换而不是重复追加。这样 checkpointer 可以自动管理完整的对话历史。

3. 为什么不用一个大 dict 乱塞？

每个字段都有明确的写入者和消费者：

这样每个节点的输入输出边界是清晰的，不会出现"这个字段到底谁写的"这种问题。

五、Step 2：实现路由节点

Router 是整个图的入口。它接收用户消息，让 LLM 判断应该走哪条路径。

打开 app/agent/nodes/router.py：

from __future__ import annotations
import json
import re
from langchain_core.messages import AIMessage, HumanMessage
from app.agent.prompts import ROUTER_SYSTEM_PROMPT
from app.agent.state import AgentState, RouteDecision, RouteType, TaskType
from app.services.llm_service import chat_completion


def _parse_json_from_response(text: str) -> dict | None:
    """
    从 LLM 响应中提取 JSON，支持多种格式：
    纯 JSON / Markdown 代码块 / 嵌套文本 / 中文引号 / 尾逗号
    """
    text = text.strip()
    if not text:
        return None

    # 预处理：中文引号 → 英文引号
    text_clean = text.replace('\u201c', '"').replace('\u201d', '"')

    # 尝试 1: 直接解析
    try:
        return json.loads(text_clean)
    except json.JSONDecodeError:
        pass

    # 尝试 2: Markdown 代码块
    match = re.search(r'```(?:json)?\s*\n(.*?)\n```', text_clean, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(1).strip())
        except json.JSONDecodeError:
            pass

    # 尝试 3: 花括号匹配
    match = re.search(r'\{.*\}', text_clean, re.DOTALL)
    if match:
        try:
             json.loads(.group())
         json.JSONDecodeError:
            

     


 () -> :
    
    messages = state.get(, [])

    
    history = messages[-:]  (messages) >   messages

    
    router_messages = [{: , : ROUTER_SYSTEM_PROMPT}]
     msg  history:
         (msg, HumanMessage):
            router_messages.append({: , : msg.content})
         (msg, AIMessage):
            router_messages.append({: , : msg.content})

    
    response = chat_completion(router_messages, temperature=, max_tokens=)

      response:
        
         {: , : }

    
    data = _parse_json_from_response(response)
     data  :
         {: , : }

    decision = RouteDecision.model_validate(data)
    route_type = decision.route_type
    task_type = decision.task_type

    
     route_type == RouteType.WEB   web_search_available:
        route_type = RouteType.RETRIEVE

     {
        : route_type.value,
        : task_type.value,
        : [AIMessage(content=)],
    }

Router 的 Prompt 长这样（prompts.py 中）：

ROUTER_SYSTEM_PROMPT = """你是一个意图路由器。根据用户的问题和对话历史，判断最合适的数据来源路径。

请分析用户的问题，选择以下路径之一：

1. **retrieve（知识库检索）**：问题与已上传的文档/知识库内容相关
   - 例："简历中的 STAR 法则怎么写？""后端岗位需要什么技术栈？"

2. **web（网络搜索）**：问题涉及最新信息、实时数据
   - 例："2026年最新的 Java 版本是什么？"

3. **direct（直接回答）**：简单的寒暄、常识性问题
   - 例："你好""你能做什么？"

同时判断任务类型：
- **qa**：普通问答
- **resume_analysis**：用户提供简历内容，希望得到分析
- **jd_analysis**：用户提供岗位描述，希望拆解岗位要求

请以 JSON 格式输出：
{"reasoning": "简要判断理由", "route_type": "retrieve|web|direct", "task_type": "qa|resume_analysis|jd_analysis"}"""

这里有几个值得注意的工程细节：

为什么 Router 用同步调用？

路由决策只需要几十个 token，不需要流式输出。同步代码更容易调试，出错时堆栈信息更清晰。

为什么要有 5 种 JSON 解析 fallback？

LLM 输出不稳定是常态。你可能遇到：

• 纯 JSON（理想情况）
• Markdown 代码块包裹：```json {...} ```
• 中文引号：{"name"："张三"}
• 尾逗号：{"skills": ["Java", "Go",],}

_parse_json_from_response 按优先级逐级尝试，最大限度避免路由失败。

为什么取最近 6 条消息？

Router 不需要完整对话历史。最近 6 条（约 3 轮对话）足以判断当前意图，同时节省 token。历史管理由 checkpointer 负责，不影响持久化。

六、Step 3：知识库检索节点

# app/agent/nodes/kb_search.py
from langchain_core.messages import HumanMessage
from app.agent.state import AgentState

_retrieval_service = None   # 由 graph.py 注入
_top_k = 5


def set_retrieval_service(service, top_k: int = 5) -> None:
    global _retrieval_service, _top_k
    _retrieval_service = service
    _top_k = top_k


def search_kb(state: AgentState) -> dict:
    """知识库检索节点"""
    messages = state.get("messages", [])
    if not messages or not _retrieval_service:
        return {"context_sources": []}

    # 取最后一条用户消息作为查询
    query = ""
    for msg in reversed(messages):
        if isinstance(msg, HumanMessage):
            query = msg.content
            break

    results = _retrieval_service.retrieve(query, top_k=_top_k)

    context_sources = []
    for item in results:
        context_sources.append({
            "content": item.get("content", ""),
            "source": item.get("source", ""),
            "score": item.get("score", 0.0),
            "page": item.get("page"),
            "type": "kb",
        })

    return {: context_sources}

依赖注入模式：_retrieval_service 不是在模块加载时创建的，而是由 graph.py 在构建图的时候通过 set_retrieval_service() 注入。这样做的好处是：

1. 节点代码不依赖全局单例
2. 测试时可以注入 mock 对象
3. 多个服务实例共享同一个检索接口

七、Step 4：结果标准化节点

检索节点返回的是结构化的 context_sources 列表，但 LLM 需要的是一段文本。normalize 节点负责这个转换：

# app/agent/nodes/normalize.py
from app.agent.state import AgentState


def normalize_kb(state: AgentState) -> dict:
    """把 KB 检索结果拼装成 LLM 可消费的文本"""
    context_sources = state.get("context_sources", [])
    kb_sources = [s for s in context_sources if s.get("type") == "kb"]

    if not kb_sources:
        return {"working_context": ""}

    parts = []
    for i, src in enumerate(kb_sources):
        label = f"【来源{i+1} - 知识库】"
        if src.get("source"):
            label += f" {src['source']}"
        parts.append(f"{label}\n{src['content']}")

    working_context = "\n\n".join(parts)
    return {"working_context": working_context}

输出格式类似：

【来源1 - 知识库】 简历写作指南.md 第3页
STAR 法则是结构化描述工作经历的方法...

【来源2 - 知识库】 后端技术栈要求.md 第1页
Java 后端工程师核心技能要求...

这个设计把 context_sources（结构化数据层）和 working_context（文本化展示层）分开了。前端可以展示结构化的引用来源，而 LLM 只看文本。

八、Step 5：生成节点

# app/agent/nodes/generate.py
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from app.agent.prompts import AGENT_SYSTEM_PROMPT, DIRECT_SYSTEM_PROMPT, WEB_AGENT_SYSTEM_PROMPT
from app.agent.state import AgentState
from app.services.llm_service import chat_completion, chat_completion_stream_async

_max_history: int = 20


def generate(state: AgentState) -> dict:
    """生成节点：拼装上下文 + 调 LLM"""
    messages = state.get("messages", [])
    working_context = state.get("working_context", "")
    route_type = state.get("route_type", "direct")
    route_str = route_type.value if hasattr(route_type, "value") else str(route_type)

    # 裁剪发送给 LLM 的历史（不影响 checkpointer 持久化）
    trimmed = messages[-_max_history:] if len(messages) > _max_history else messages

    # 根据 route_type 选择系统 Prompt
    system_map = {
        "web": WEB_AGENT_SYSTEM_PROMPT,
        "retrieve": AGENT_SYSTEM_PROMPT,
        "direct": DIRECT_SYSTEM_PROMPT,
    }
    system_prompt = system_map.get(route_str, DIRECT_SYSTEM_PROMPT)

    # 构造消息列表
    llm_messages = [{"role": "system", "content": system_prompt}]
    for msg in trimmed:
        if isinstance(msg, HumanMessage):
            llm_messages.append({"role": "user", "content": msg.content})
        elif (msg, AIMessage)   msg.content.startswith():
            llm_messages.append({: , : msg.content})

    
     working_context:
        last_user_msg = 
         msg  (trimmed):
             (msg, HumanMessage):
                last_user_msg = msg.content
                
        context_block = 
         llm_messages  llm_messages[-][] == :
            llm_messages[-] = {: , : context_block}

    answer = chat_completion(llm_messages)
     {: answer, : [AIMessage(content=answer)]}

这里有两个细节值得注意：

1. 路由决策消息被过滤了

elif isinstance(msg, AIMessage) and not msg.content.startswith("[路由决策]"):

Router 节点会往 messages 里塞一条 [路由决策] 走retrieve路径 的消息。这条消息是给 checkpointer 记录用的，不应该发给 LLM（否则会干扰回答质量）。

2. 上下文注入方式

检索到的上下文不是作为 system message 注入的，而是替换最后一条用户消息：

参考内容：
【来源1 - 知识库】简历写作指南.md
STAR 法则是...

---
用户问题：简历中的 STAR 法则怎么写？

这种方式比单独添加 system message 效果更好，因为 LLM 能更自然地理解"用户问题是基于这些参考内容提出的"。

九、Step 6：组装主图

所有节点都写好了，现在用 LangGraph 把它们串起来：

# app/agent/graph.py
from functools import partial
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from app.agent.state import AgentState
from app.agent.nodes.router import route_query
from app.agent.nodes.kb_search import search_kb, set_retrieval_service
from app.agent.nodes.web_search import search_web, set_web_search_service
from app.agent.nodes.normalize import normalize_kb, normalize_web
from app.agent.nodes.generate import generate_streaming_node, set_max_history

_checkpointer = MemorySaver()


def _route_decision(state: AgentState) -> str:
    """条件边：根据 task_type + route_type 决定走哪条路径"""
    task_type = state.get("task_type", "qa")
    route_type = state.get("route_type", "direct")

    task_str = task_type.value if hasattr(task_type, "value") else str(task_type)
    route_str = route_type.value if hasattr(route_type, "value") else str(route_type)

    if task_str == "resume_analysis":
        return "resume_analysis"
    if task_str == "jd_analysis":
        return "jd_analysis"

    return route_str   # "retrieve" | "web" | "direct"


def () -> StateGraph:
    

    
    set_retrieval_service(retrieval_service, top_k=settings.top_k)
    set_web_search_service(web_search_service)
    set_max_history(settings.agent_max_history)

    builder = StateGraph(AgentState)

    
    builder.add_node(, partial(route_query, web_search_available=web_search_service.is_available))
    builder.add_node(, search_kb)
    builder.add_node(, search_web)
    builder.add_node(, normalize_kb)
    builder.add_node(, normalize_web)
    builder.add_node(, generate_streaming_node)

    
    builder.add_edge(START, )

    
    builder.add_conditional_edges(
        ,
        _route_decision,
        {
            : ,
            : ,
            : ,
            : ,
            : ,
        },
    )

    
    builder.add_edge(, )
    builder.add_edge(, )
    builder.add_edge(, )
    builder.add_edge(, )

    
    builder.add_edge(, END)

    
     builder.(checkpointer=_checkpointer)

这段代码的核心是 add_conditional_edges。_route_decision 返回一个字符串键，LangGraph 根据这个键把执行流导向对应的目标节点。

Checkpointer 的作用：MemorySaver 在进程内存中存储每个 thread（session）的 state 快照。每次 ainvoke 或 astream 执行时，checkpointer 自动加载该 thread 的历史 state，执行完毕后自动保存新 state。这就是多轮对话能"记住"之前聊过什么的原因。

十、Step 7：API 层

最后，用 FastAPI 把图暴露为 HTTP 接口：

# app/api/agent.py
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage
from app.schemas.agent import AgentChatRequest, AgentChatResponse

router = APIRouter(prefix="/agent", tags=["Agent"])
_agent_graph = None


def set_agent_graph(graph):
    global _agent_graph
    _agent_graph = graph


@router.post("/chat", response_model=AgentChatResponse)
async def agent_chat(request: AgentChatRequest):
    """非流式对话接口"""
    session_id = request.session_id or uuid.uuid4().hex
    config = {"configurable": {"thread_id": session_id}}

    input_state = {
        "messages": [HumanMessage(content=request.question)],
        "context_sources": [],
        "working_context": "",
        "final_answer": "",
    }

    result = await _agent_graph.ainvoke(input_state, config=config)

    return AgentChatResponse(
        answer=result.get("final_answer", ""),
        session_id=session_id,
        route_type=result.get("route_type", "direct"),
    )


@router.post("/chat/stream")
async def agent_chat_stream(request: AgentChatRequest):
    """流式对话接口（SSE）"""

    async  ():
        config = {: {: session_id}}
        input_state = {
            : [HumanMessage(content=request.question)],
            : [],
            : ,
            : ,
        }

          mode, payload  _agent_graph.astream(
            input_state, config=config,
            stream_mode=[, ],
        ):
             mode == :
                 payload.get() == :
                     
             mode == :
                   payload:
                    route = payload[].get()
                     
                   payload    payload:
                    sources = payload.get(, payload.get(, {})).get(, [])
                     

         

    session_id = request.session_id  uuid.uuid4().

     StreamingResponse(
        event_generator(),
        media_type=,
        headers={: , : },
    )

SSE 事件流是前端实时显示的关键。前端通过 EventSource 或 fetch + ReadableStream 监听：

data: {"type": "route", "route": "retrieve"}     ← 走了检索路径
data: {"type": "sources", "sources": [...]}       ← 检索结果
data: {"type": "token", "content": "STAR"}        ← 流式 token（多次）
data: {"type": "token", "content": "法则是..."}
...
data: {"type": "done", "session_id": "abc123"}    ← 完成

十一、启动入口

最后，main.py 把所有东西组装起来：

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.agent.graph import build_agent_graph, get_checkpointer
from app.api.agent import router as agent_router, set_agent_graph, set_checkpointer
from app.services.retrieval_service import RetrievalService
from app.services.web_search_service import WebSearchService
from app.core.config import get_settings

settings = get_settings()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时构建图并注入依赖
    retrieval_service = RetrievalService(settings.knowledge_base_dir)
    web_search_service = WebSearchService(settings)

    graph = build_agent_graph(retrieval_service, web_search_service, settings)
    set_agent_graph(graph)
    set_checkpointer(get_checkpointer())

    yield


app = FastAPI(title="ResumeAgent", lifespan=lifespan)
app.include_router(agent_router)

运行：

uvicorn main:app --reload --port 8000

打开 http://localhost:8000/docs 就能看到所有 API 接口了。

本篇小结

到这里，我们已经搭好了系统的骨架：

1. ✅ 状态模型——每个字段有明确的写入者和消费者
2. ✅ 路由节点——LLM 结构化输出 + 5 种 JSON 解析 fallback
3. ✅ QA 路径——知识库检索 → 标准化 → 生成，支持 retrieve / web / direct 三条支路
4. ✅ 主图编译——条件边 + checkpointer 实现多轮对话持久化
5. ✅ SSE 流式接口——前端实时展示 token

但这只是 QA 路径。真正的难点在于简历分析和 JD 分析——它们各自有独立的多步处理链路，而且需要在流式输出的同时保持子图的状态管理。

下一篇，我会手把手教你：

• 如何用 LangGraph 子图拆分复杂分析路径
• 如何在子图内实现流式 token 输出
• 如何让 API 层只做事件翻译，不编排节点顺序
• 以及开发过程中踩过的 4 个大坑和解决方案

下一篇见。

完整代码仓库：[https://github.com/SkylarkLiu/ResumeAgent]

个人技术博客：[https://superskylark.icu]

如果觉得有帮助，欢迎点赞收藏，下篇更精彩 👋