LangGraph 多 Agent 编排实战：5 个 Agent 如何协作完成职业规划分析 · devcfg

0. 系列闭环（不公开源码也能跟读）

端到端链路：Vue 前端 → api/routes/chat.py → Guide 多轮 SSE → run_analysis_pipeline（解析→分析→匹配→报告）→ tools/pdf_exporter PDF。
本篇：第 2/17 篇 · 流水线 · 五 Agent 节点

阶段	用户可见	代码入口	对应篇
建会话	欢迎语	POST /api/sessions	09
多轮对话	SSE 流式	chat/stream → run_guide_single_turn	06, 14
信息充分	开始分析	_run_analysis_background	05, 07
履历解析	进度 30%	run_resume_parser	12
画像/RIASEC	进度 50%	run_profile_analyzer	03, 13
职业匹配	进度 70%	run_career_matcher	02
报告	进度 90%	run_reporter	11
下载 PDF	文件	GET …/report/pdf	11, 15

	说明
读本篇前	第 01 篇架构图
读完本篇	对照 workflow.py 说出每个 node 的输入/输出字段
下一环	第 04 篇：iCanWorkflowState 字段含义（第 3 篇）

全系列闭环索引：SERIES-LOOP.md

一、为什么需要多 Agent

一个常见的误解是：给 LLM 一个超级 Prompt 就能完成所有任务。但实际操作中你会发现：

单 Prompt 方案的问题：

• 上下文窗口浪费：把所有任务的 Prompt 塞进一次调用，有效信息被稀释
• 输出质量下降：LLM 同时做解析、分析、匹配、撰写，每件事都做不精
• 无法复用：换一个场景（比如只做简历解析），整个 Prompt 要重写
• 调试困难：输出有问题，不知道是哪个环节出错的

多 Agent 方案的优势：

• 每个 Agent 专注一件事，Prompt 更精准
• 状态在 Agent 间传递，上游的输出是下游的输入
• 单个 Agent 可以独立测试和优化
• 某个 Agent 失败不影响其他 Agent 的设计

二、5 个 Agent 的职责划分

flowchart LR
  G[guide_node] --> P[resume_parser_node]
  P --> A[profile_analyzer_node]
  A --> M[career_matcher_node]
  M --> R[reporter_node]
  G -.信息不足.-> G

Guide — 对话引导（agents/guide.py）

职责：通过多轮对话收集用户的职业信息。

它有自己的内部 StateGraph（5 个节点）：

welcome → assess_need → collect_basic_info → dig_deeper → check_sufficiency
                                                              ↓
                                                    should_continue()
                                                    ↙           ↘
                                          dig_deeper           END

• welcome：生成欢迎语，了解用户来意
• assess_need：判断核心需求（职业规划/转型/技能提升）
• collect_basic_info：收集教育、工作年限、岗位等
• dig_deeper：追问工作偏好、价值观、性格特点
• check_sufficiency：LLM 判断信息是否充分，不充分则回到 dig_deeper

关键设计：check_sufficiency 通过条件路由 should_continue() 决定走向。内部最多循环 8 次，防止无限循环。

ResumeParser — 履历解析（agents/resume_parser.py）

职责：将用户的自然语言描述转为结构化 JSON 数据。顶层 workflow.py 的 resume_parser_node 会把 conversation_history 里所有 user 发言与 raw_input 拼成 combined_text，再调 run_resume_parser()。

输入是一段拼接了所有对话内容的文本，输出是标准化的结构化画像：

{
    "basic_info": {"name": "...", "age": "...", "education": "..."},
    "work_experience": [{"company": "...", "position": "...", "duration": "..."}],
    "skill_set": {"technical_skills": [...], "soft_skills": [...], "tools": [...]},
    "career_progression": {"total_years": 5, "industries": [...], "career_path": "..."}
}

技术难点：LLM 输出的 JSON 格式不稳定。agents/resume_parser.py 配合 llm/parsers.py 的四层降级解析 + llm/providers.py 的 invoke_llm_with_json（response_format={"type": "json_object"}）强制 JSON 模式；模型走 get_light_model()。

ProfileAnalyzer — 个人分析（agents/profile_analyzer.py）

职责：基于结构化画像进行五维度深度分析。

五个维度：

1. 能力模型：硬实力/软技能/学习力/创新力/领导力（0-10 分）
2. 工作风格：决策方式/协作偏好/节奏偏好/沟通风格
3. 性格特质：大五人格五维度评分
4. 职业价值观：8 个维度排序（物质回报/成长/平衡/影响力/自主性/稳定/创新/人际）
5. 霍兰德 RIASEC：R/I/A/S/E/C 六维度评分 + Holland Code

技术亮点：子图 create_profile_analyzer_graph() 在 load_profile 之后顺序执行能力 → 风格 → 性格 → 价值观 → analyze_riasec → 优短板 → synthesize_profile（注释里写「并行」是设计意图，当前边是线性的）。霍兰德打分见第 3、13 篇。

CareerMatcher — 职业匹配（agents/career_matcher.py）

职责：基于个人画像推荐三级职业路径。

三级推荐策略：

级别	策略	匹配度	说明
第一级	纵向深耕	80-95%	在当前行业继续深入
第二级	横向拓展	60-80%	迁移到相关领域
第三级	转型探索	40-60%	基于个人特质的全新方向

每级推荐包含：目标岗位、技能差距、市场前景（需求/薪资/趋势）、时间线。

Reporter — 报告输出（agents/reporter.py）

职责：将所有分析结果整合为结构化 Markdown 报告。

报告结构：

1. 个人画像总览
2. 五维度深度分析
3. 职业方向推荐（三级）
4. 行动建议（短/中/长期）
5. 市场洞察

三、StateGraph 的串联实现

实现文件：workflow.py（顶层唯一编排入口，见 SOURCE-ACCURACY.md）。

顶层工作流 create_workflow()

# workflow.py — 节选
from langgraph.graph import StateGraph, END
from ican.core.state import iCanWorkflowState
from ican.agents.guide import run_guide_agent
# ... resume_parser / profile_analyzer / career_matcher / reporter ...

graph = StateGraph(iCanWorkflowState)
graph.add_node("guide_node", guide_node)
graph.add_node("resume_parser_node", resume_parser_node)
graph.add_node("profile_analyzer_node", profile_analyzer_node)
graph.add_node("career_matcher_node", career_matcher_node)
graph.add_node("reporter_node", reporter_node)

graph.set_entry_point("guide_node")
graph.add_conditional_edges(
    "guide_node",
    route_after_guide,
    {
        "guide_node": "guide_node",
        "resume_parser_node": "resume_parser_node",
    },
)
graph.add_edge("resume_parser_node", "profile_analyzer_node")
graph.add_edge("profile_analyzer_node", "career_matcher_node")
graph.add_edge("career_matcher_node", "reporter_node")
graph.add_edge("reporter_node", END)
compiled_graph = graph.compile()

外层 guide_node：内外层状态桥接

guide_node 负责把 iCanWorkflowState（core/state.py）映射到内层 GuideState，调用 run_guide_agent()，再把 is_info_sufficient 转成外层的 needs_more_info：

# workflow.py — guide_node 核心逻辑
guide_state = create_initial_guide_state()
guide_state["conversation_history"] = conversation_history
guide_result = await run_guide_agent(guide_state)

return {
    "conversation_history": updated_history,
    "current_agent": "guide",
    "needs_more_info": not guide_result.get("is_info_sufficient", False),
}

外层路由 route_after_guide

与内层 agents/guide.py 的 should_continue（loop_count >= 8）配合，外层还有一道保险：user_msg_count >= 3 时强制进入解析，避免无用户交互时死循环：

# workflow.py — route_after_guide
def route_after_guide(state: iCanWorkflowState) -> str:
    if not state.get("needs_more_info", True):
        return "resume_parser_node"
    user_msg_count = len([m for m in state.get("conversation_history", [])
                          if m.get("role") == "user"])
    if user_msg_count >= 3:
        return "resume_parser_node"
    return "guide_node"

生产环境：run_analysis_pipeline

前端 SSE 对话在信息充分后通常不走完整 ainvoke(create_workflow())，而是 workflow.py 的 run_analysis_pipeline()：跳过 Guide，直接串四个 Agent，并在 Ollama 不可用时降级为规则引擎报告。入口在 api/routes/chat.py 与 api/routes/report_gen.py。

节点间的数据传递

每个节点函数的模式：

# workflow.py — profile_analyzer_node 模式
async def profile_analyzer_node(state: iCanWorkflowState) -> dict:
    structured_profile = state.get("structured_profile", {})
    analyzer_state: ProfileAnalysisState = {"structured_profile": structured_profile}
    analyzer_result = await run_profile_analyzer(analyzer_state)

    complete_profile = {
        "structured_profile": structured_profile,
        "ability_model": analyzer_result.get("ability_model", {}),
        "riasec_scores": analyzer_result.get("riasec_scores", {}),
        "strengths": analyzer_result.get("strengths", []),
        # work_style / personality_traits / career_values ...
    }
    return {"personal_profile": complete_profile, "current_agent": "profile_analyzer"}

关键点：节点只返回需要更新的字段，LangGraph 自动合并到全局状态中。

四、与 CrewAI / AutoGen 的对比

维度	LangGraph	CrewAI	AutoGen
编排模型	有向图（DAG）	流程/层级	对话轮次
状态管理	TypedDict + Reducer	共享 Memory	消息历史
条件路由	✅ 原生支持	⚠️ 需要自定义	❌ 不支持
循环控制	✅ 条件边 + recursion_limit	⚠️ 有限支持	✅ 对话循环
可视化	✅ 导出图结构	❌	❌
学习曲线	中等	低	低
适用场景	复杂工作流	简单任务编排	多模型对话

选择建议：

• 如果你的 Agent 之间有条件路由和循环 → LangGraph
• 如果只是简单的任务分配 → CrewAI
• 如果需要多模型对话/辩论 → AutoGen

五、踩坑经验

坑 1：Reducer 字段层级搞混

外层 iCanWorkflowState（core/state.py）用 workflow_messages: Annotated[list[str], operator.add] 累积工作流日志；内层 GuideState 才用 messages: Annotated[list[str], operator.add] 累积 AI 回复。若在顶层误用 messages，LangGraph 不会按预期合并。

# core/state.py — 外层
workflow_messages: Annotated[list[str], operator.add]

# core/state.py — 内层 GuideState
messages: Annotated[list[str], operator.add]

坑 2：循环不可控

Guide Agent 可能无限循环。解决方案是双层限制：

• 内层：should_continue() 限制最多 8 次迭代
• 外层：route_after_guide() 限制最多 3 轮用户消息

坑 3：节点异常与空画像

workflow.py 各节点独立 try-except，失败时返回空 dict 或占位报告（如 reporter_node 的「报告生成失败，请稍后重试」），避免整图崩溃。但若 structured_profile 为空仍进入 profile_analyzer_node，下游 RIASEC 会全零分——需在 Guide 或 Parser 层保证输入厚度。

坑 4：实现形态是函数 + 子图，不是 Agent 类

五个节点均为 async def xxx_node + run_xxx() 入口，不存在 GuideAgent Python 类；对照源码时搜 agents/guide.py 的 run_guide_agent，勿按 CrewAI 的 Role 类去理解。

六、小结（iCan 编排要点）

• 顶层只有 workflow.py 的 create_workflow()，节点名固定：guide_node → route_after_guide → 四段线性链。
• 每个节点做外层 TypedDict → 内层 TypedDict → run_xxx → 映射回外层 四步，字段定义见 core/state.py。
• 双层循环上限：内层 should_continue（loop_count >= 8）+ 外层 route_after_guide（user_msg_count >= 3）。
• 线上 SSE 路径用 run_guide_chat + run_analysis_pipeline，与 CLI 一次性 run_workflow 不同。

下一篇：霍兰德 RIASEC 在 agents/profile_analyzer.py 的工程实现，以及 OpenAI 兼容 API（DeepSeek 等为 .env 部署示例）下的结构化打分。

---

← 返回 iCan 专题