基于 LangGraph 的智能简历优化系统，通过 AI Agent 工作流自动解析、优化和评估简历，帮助求职者提升简历与目标岗位的匹配度。
LangGraph 和 Langchain教学:https://docs.langchain.com/oss/python/langchain/overview
LangSmith网站:https://www.langchain.com/langsmith/observability
Python 的版本3.13.2
# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 安装依赖
pip install -r requirements.txt修改 .env 文件，填入以下配置：
# 必需：DeepSeek API Key（用于 Parser, Editor, Evaluator）
DEEPSEEK_API_KEY=your_deepseek_api_key
# 必需：硅基流动 API Key（用于 Summary Agent）
SILICONFLOW_API_KEY=your_siliconflow_api_key
# 可选：LangSmith 追踪（用于调试和监控）
LANGCHAIN_API_KEY=your_langchain_api_key
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=yukesong-resume-optimizeruvicorn app.main:app --reload访问 http://localhost:8000/docs 查看交互式 API 文档。
app/
├── main.py # FastAPI 应用入口，配置 CORS 和路由
├── api/
│ └── routes.py # API 路由定义：/resume/optimize
├── agents/ # AI Agent 实现
│ ├── resume_parser.py # 解析 Agent：PDF → 结构化数据
│ ├── resume_editor.py # 编辑 Agent：根据 JD 优化简历
│ ├── hm_evaluator.py # 评估 Agent：评估匹配度并给出反馈
│ └── summary_agent.py # 摘要 Agent：生成优化摘要
├── workflows/
│ └── resume_workflow.py # LangGraph 工作流编排（核心逻辑）
├── models/ # Pydantic 数据模型
│ ├── schemas.py # API 请求/响应模型
│ └── agent_outputs.py # Agent 输出结构化模型
├── services/
│ └── resume_service.py # 业务逻辑层，调用工作流
├── tools/
│ └── pdf_parser.py # PDF 文本提取工具
├── prompts/ # 提示词模板
│ ├── parser_prompts.py # 解析提示词
│ ├── editor_prompts.py # 编辑提示词
│ ├── hm_prompts.py # 评估提示词
│ └── summary_prompts.py # 摘要生成提示词
└── core/
└── config.py # 配置管理（API Key、模型等）
- API 接收 (
api/routes.py) → 接收 PDF 文件和职位描述（JD） - 服务层 (
services/resume_service.py) → 调用工作流处理 - 工作流编排 (
workflows/resume_workflow.py) → 使用 LangGraph 编排多个 Agent - Agent 执行 → 解析 → 编辑 → 评估 → [迭代优化?]
- 返回结果 → 优化建议、反馈和最终评估
workflows/resume_workflow.py- 工作流定义，核心逻辑，使用 LangGraph 编排 Agentagents/*.py- 每个文件实现一个 Agent，负责特定任务models/agent_outputs.py- Agent 输出的结构化数据模型（使用 Pydantic）core/config.py- 所有配置集中管理，包括 API Key、模型参数等
系统使用 LangGraph 构建了一个多 Agent 协作的工作流，实现简历的自动优化和迭代改进。
graph TD
A[接收请求] --> B[解析简历]
B --> C[编辑优化]
C --> D[评估匹配度]
D --> E{是否通过?}
E -->|通过| F[生成摘要]
E -->|未通过且可继续| C
E -->|达到最大次数| F
F --> G[格式化结果]
G --> H[返回结果]
H --> I[结束]
style A fill:#e1f5ff
style B fill:#fff4e1
style C fill:#fff4e1
style D fill:#fff4e1
style E fill:#ffe1f5
style F fill:#ffe1f5
style G fill:#e1ffe1
style H fill:#e1f5ff
流程图节点：接收请求
对应代码位置：
- API 路由：
app/api/routes.py::optimize_resume() - 服务层：
app/services/resume_service.py::process_resume() - 工作流：
app/workflows/resume_workflow.py::run_workflow()
功能：接收 PDF 文件和职位描述，初始化工作流状态
流程图节点：解析简历
对应代码位置：app/workflows/resume_workflow.py::parse_resume()
功能：使用 ResumeParserAgent 将 PDF 文件解析为结构化 JSON 数据
输出：state["resume_blocks"] 包含所有简历模块（summary、education、work_experience、projects、skills、certificates、languages）
流程图节点：编辑优化
对应代码位置：app/workflows/resume_workflow.py::edit_resume()
功能：使用 ResumeEditorAgent 根据 JD 和反馈优化简历内容
输入：原始简历、职位描述、评估反馈（首次为 None）
输出：state["optimized_blocks"] 优化后的简历数据，state["iteration_count"] 迭代次数 +1
流程图节点：评估匹配度
对应代码位置：app/workflows/resume_workflow.py::evaluate_resume()
功能：使用 HMEvaluatorAgent 评估优化后的简历与 JD 的匹配度
输出：
state["hm_approved"]: 是否通过评估（bool）state["hm_feedback"]: 评估反馈文本（包含优势、不足、改进建议等）
流程图节点：是否通过?
对应代码位置：app/workflows/resume_workflow.py::should_continue()
判断逻辑：
- 通过 (
hm_approved == True) → 生成摘要 - 未通过且可继续 (
iteration_count < MAX_ITERATIONS) → 返回编辑优化（循环） - 达到最大次数 (
iteration_count >= MAX_ITERATIONS) → 生成摘要
配置：最大迭代次数 MAX_ITERATIONS（默认 5 次）
流程图节点：生成摘要
对应代码位置：app/workflows/resume_workflow.py::generate_summary()
功能：使用 SummaryAgent 分析原始简历和优化后简历的差异，生成按模块分类的优化摘要
输出：state["optimization_summary"] 包含 6 个模块的优化说明（education、work_experience、projects、skills、certificates、languages）
流程图节点：格式化结果
对应代码位置：app/workflows/resume_workflow.py::format_result()
功能：将工作流状态中的结果整理成最终返回格式，包含：
approved: 是否通过评估iteration_count: 迭代次数resume_blocks: 原始简历数据optimized_blocks: 优化后简历数据hm_feedback: HR 评估反馈summary: 按模块分类的优化摘要（education、work_experience、projects、skills、certificates、languages）
流程图节点：返回结果
对应代码位置：
- 工作流：
app/workflows/resume_workflow.py::run_workflow() - API：
app/api/routes.py::optimize_resume()
响应内容：
approved: 是否通过评估iteration_count: 迭代次数resume_blocks: 原始简历数据optimized_blocks: 优化后简历数据hm_feedback: HR 评估反馈summary: 按模块分类的优化摘要（6 个模块）
系统支持迭代优化：如果简历未通过评估，会将评估反馈传递给编辑 Agent，进行下一轮优化，最多迭代 5 次（可在配置中修改）。
优化简历接口。
请求参数：
resume_pdf(file, required): PDF 格式的简历文件job_description(string, required): 职位描述，10-10000 字符
响应字段：
approved: 是否通过评估iteration_count: 优化迭代次数resume_blocks: 原始简历数据optimized_blocks: 优化后简历数据hm_feedback: HR 评估反馈summary: 按模块分类的优化摘要（education、work_experience、projects、skills、certificates、languages）
详细 API 文档：请参考 API_RESPONSE_FORMAT.md
LangGraph 是 LangChain 的一个扩展，用于构建有状态的多 Agent 工作流。它特别适合需要条件分支、循环和状态管理的复杂 AI 应用。
工作流的状态是一个字典，包含所有节点之间共享的数据。
from typing import TypedDict
class WorkflowState(TypedDict):
resume_path: str
job_description: str
resume_blocks: Dict[str, Any] | None
optimized_blocks: Dict[str, Any] | None
hm_feedback: str | None
hm_approved: bool
iteration_count: int
optimization_summary: Dict[str, str] | None # Summary by module
final_result: Dict[str, Any] | None要点：
- 使用
TypedDict定义状态结构，提供类型提示 - 所有节点都可以读取和修改状态
- 状态在节点之间自动传递
节点是工作流中的执行单元，通常是一个异步函数。
async def parse_resume(state: WorkflowState) -> WorkflowState:
# 读取状态
resume_path = state["resume_path"]
# 执行任务
resume_blocks = await parser.parse_resume(resume_path)
# 更新状态
state["resume_blocks"] = resume_blocks
# 返回更新后的状态
return state要点：
- 节点函数接收
state作为参数 - 必须返回更新后的
state - 可以是异步函数（推荐）或同步函数
图用于定义节点之间的连接关系。
from langgraph.graph import StateGraph, END
# 创建图
workflow = StateGraph(WorkflowState)
# 添加节点
workflow.add_node("parse_resume", parse_resume)
workflow.add_node("edit_resume", edit_resume)
workflow.add_node("evaluate_resume", evaluate_resume)
# 设置入口点
workflow.set_entry_point("parse_resume")
# 添加边（顺序执行）
workflow.add_edge("parse_resume", "edit_resume")
workflow.add_edge("edit_resume", "evaluate_resume")条件边根据状态决定下一步执行哪个节点。
def should_continue(state: WorkflowState) -> str:
if state["hm_approved"]:
return "approve"
if state["iteration_count"] >= 5:
return "reject"
return "continue"
# 添加条件边
workflow.add_conditional_edges(
"evaluate_resume", # 源节点
should_continue, # 条件函数
{
"continue": "edit_resume", # 如果返回 "continue"，跳转到 edit_resume
"approve": "format_result", # 如果返回 "approve"，跳转到 format_result
"reject": "format_result" # 如果返回 "reject"，跳转到 format_result
}
)要点：
- 条件函数接收
state，返回一个字符串（路由键） - 路由键必须匹配
add_conditional_edges中的映射
# 编译图
compiled_workflow = workflow.compile()
# 执行工作流
initial_state: WorkflowState = {
"resume_path": "path/to/resume.pdf",
"job_description": "职位描述...",
"resume_blocks": None,
# ... 其他字段
}
# 同步执行
result = compiled_workflow.invoke(initial_state)
# 异步执行（推荐）
result = await compiled_workflow.ainvoke(initial_state)from langgraph.graph import StateGraph, END
from typing import TypedDict
class MyState(TypedDict):
count: int
result: str | None
def increment(state: MyState) -> MyState:
state["count"] += 1
return state
def check_count(state: MyState) -> str:
if state["count"] >= 5:
return "done"
return "continue"
workflow = StateGraph(MyState)
workflow.add_node("increment", increment)
workflow.set_entry_point("increment")
workflow.add_conditional_edges(
"increment",
check_count,
{
"continue": "increment", # 循环
"done": END
}
)
compiled = workflow.compile()
result = compiled.invoke({"count": 0, "result": None})- 顺序执行：使用
add_edge - 条件分支：使用
add_conditional_edges - 循环：条件边指向自身
- 并行执行：多个节点指向同一个节点（LangGraph 会自动处理）
- 使用
LangSmith追踪工作流执行过程 - 在节点中添加日志输出
- 使用
workflow.get_graph().draw_mermaid()可视化工作流图
Pydantic 用于数据验证和设置管理，通过类型注解自动验证。
from pydantic import BaseModel, Field
from typing import Optional, List
class EducationItem(BaseModel):
institution: str = Field(..., description="学校名称")
degree: str = Field(..., description="学位")
gpa: Optional[str] = Field(None, description="GPA（可选）")
courses: List[str] = Field(default_factory=list) # 使用 default_factory 避免共享可变对象- Agent 输出模型 (
models/agent_outputs.py): 确保 Agent 返回结构一致 - API 模型 (
models/schemas.py): 自动验证输入，生成文档 - 工作流状态 (
workflows/resume_workflow.py): 使用TypedDict定义状态
- 在
app/agents/创建新文件，例如custom_agent.py：
from langchain_openai import ChatOpenAI
from app.models.agent_outputs import ResumeBlocks
class CustomAgent:
def __init__(self, llm: ChatOpenAI):
self.llm = llm
async def process(self, state: dict) -> dict:
# 实现你的逻辑
return result- 在
workflows/resume_workflow.py中添加节点：
from app.agents.custom_agent import CustomAgent
# 在 create_workflow() 中
custom_agent = CustomAgent(llm)
async def custom_node(state: WorkflowState) -> WorkflowState:
result = await custom_agent.process(state)
state["custom_field"] = result
return state
workflow.add_node("custom_node", custom_node)
workflow.add_edge("previous_node", "custom_node")编辑 app/prompts/ 目录下的对应文件：
parser_prompts.py- 解析提示词editor_prompts.py- 编辑提示词hm_prompts.py- 评估提示词summary_prompts.py- 摘要生成提示词
- Agent 输出模型：编辑
app/models/agent_outputs.py - API 模型：编辑
app/models/schemas.py - 工作流状态：编辑
workflows/resume_workflow.py中的WorkflowState
编辑 app/core/config.py，添加或修改配置项：
class Settings:
# 添加新配置
NEW_CONFIG: str = os.getenv("NEW_CONFIG", "default_value")在 app/core/config.py 中可以修改：
DEEPSEEK_API_KEY: DeepSeek API Key（必需，用于 Parser, Editor, Evaluator）DEEPSEEK_BASE_URL: DeepSeek API 地址（默认：https://api.deepseek.com）DEEPSEEK_MODEL: DeepSeek 模型名称（默认：deepseek-chat）SILICONFLOW_API_KEY: 硅基流动 API Key（必需，用于 Summary Agent）SILICONFLOW_BASE_URL: 硅基流动 API 地址（默认：https://api.siliconflow.cn/v1）SUMMARY_MODEL: Summary Agent 使用的模型（默认：Qwen/Qwen3-8B，通过硅基流动调用）DEFAULT_MODEL: 其他 Agent 使用的模型（默认：deepseek-chat，通过 DeepSeek API 调用）MAX_TOKENS: 最大生成 token 数（默认：4096）MAX_ITERATIONS: 最大迭代次数（默认：5）MAX_UPLOAD_SIZE: 最大上传文件大小（默认：10MB）UPLOAD_DIR: 上传文件目录（默认：uploads）
运行测试脚本：
python test.py