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

端到端链路：Vue 前端 → api/routes/chat.py → Guide 多轮 SSE → run_analysis_pipeline（解析→分析→匹配→报告）→ tools/pdf_exporter PDF。
本篇：第 5/17 篇 · 控制环 · 条件边

阶段 用户可见 代码入口 对应篇
建会话 欢迎语 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
说明
读本篇前 第 04 篇状态字段、第 06 篇 Guide 子图
读完本篇 写出 route_after_guide 与 should_continue 的退出条件
下一环 第 07 篇：路由失败时的容错（第 6 篇）

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

一、要解决什么问题

iCan 的对话引导阶段需要「信息不够就继续问、够了就进分析流水线」。如果用普通 Python if-else 串函数，很难表达「回到上一步再跑一轮」的循环，也没有 LangGraph 自带的执行追踪和 recursion_limit 兜底。

项目在两层 StateGraph 里用 add_conditional_edges 解决这件事：

  1. 内层（agents/guide.py 的 create_guide_graph()）：Guide 子图在 check_sufficiency 后决定继续 dig_deeper 还是 END。
  2. 外层（workflow.py 的 create_workflow()）：顶层图在 guide_node 后根据 needs_more_info 决定回到 guide_node 还是进入 resume_parser_node。

另外，线上 HTTP 对话走的是 run_guide_chat 单轮模式，并不每次请求都跑完整顶层图——这一点和 CLI 的 run_workflow 不同，下文会单独说明。

二、实现位置

层级 文件 关键符号
内层子图 agents/guide.py should_continue、create_guide_graph、run_guide_agent
外层编排 workflow.py guide_node、route_after_guide、create_workflow
API 单轮 workflow.py + api/routes/chat.py run_guide_chat → run_guide_single_turn
状态字段 core/state.py needs_more_info、conversation_history

Guide 条件路由

三、LangGraph 条件边 API

核心写法在 workflow.py 的 create_workflow()：

1
2
3
4
5
6
7
8
graph.add_conditional_edges(
    "guide_node",
    route_after_guide,
    {
        "guide_node": "guide_node",
        "resume_parser_node": "resume_parser_node",
    },
)

路由函数只读 state、只返回字符串 key，不发起副作用——这是 LangGraph 推荐的模式，便于调试和可视化。

四、内层：should_continue 与 Guide 子图

agents/guide.py 中 Guide 子图结构：

1
2
3
4
5
welcome → assess_need → collect_basic_info → dig_deeper → check_sufficiency
                                                              ↓
                                                     should_continue()
                                                     ↙           ↘
                                              dig_deeper           END

should_continue 的逻辑：

1
2
3
4
5
6
7
8
9
10
def should_continue(state: GuideState) -> str:
    is_sufficient = state.get("is_info_sufficient", False)
    messages_list = state.get("messages", [])
    loop_count = len([m for m in messages_list if m])

    if is_sufficient:
        return "handoff"
    if loop_count >= 8:
        return "handoff"
    return "dig_deeper"

注册方式：

1
2
3
4
5
graph.add_conditional_edges(
    "check_sufficiency",
    should_continue,
    {"dig_deeper": "dig_deeper", "handoff": END},
)

run_guide_agent 编译子图时设 recursion_limit=15，比外层的 50 更紧，防止子图在无用户输入时自旋。

五、外层：guide_node 与 route_after_guide

guide_node：把子图结果写回顶层 state

workflow.py 的 guide_node 做三件事：

  1. 把 raw_input 追加进 conversation_history；
  2. 调用 run_guide_agent(guide_state)；
  3. 用 needs_more_info = not is_sufficient 写回顶层 state。
1
2
3
4
5
6
7
8
9
async def guide_node(state: iCanWorkflowState) -> dict:
    # ...
    guide_result = await run_guide_agent(guide_state)
    is_sufficient = guide_result.get("is_info_sufficient", False)
    return {
        "conversation_history": updated_history,
        "current_agent": "guide",
        "needs_more_info": not is_sufficient,
    }

若子图抛错，节点 catch 后仍设 needs_more_info: True，避免误进分析阶段。

route_after_guide：信息充分或强制推进

1
2
3
4
5
6
7
8
9
10
def route_after_guide(state: iCanWorkflowState) -> str:
    needs_more_info = state.get("needs_more_info", True)
    conversation_history = state.get("conversation_history", [])
    user_msg_count = len([m for m in conversation_history if m.get("role") == "user"])

    if not needs_more_info:
        return "resume_parser_node"
    if user_msg_count >= 3:
        return "resume_parser_node"
    return "guide_node"

决策顺序：先信 LLM/子图的充分性判断，再数用户轮次，最后才继续循环。路由本身也有 try/except，异常时默认 resume_parser_node，避免卡死在 guide 环。

完整顶层路径（create_workflow 注释）：

1
2
3
4
start → guide_node → route_after_guide
    → (信息不足且轮次<3) → guide_node [循环]
    → (信息充分或轮次≥3) → resume_parser_node → profile_analyzer_node
    → career_matcher_node → reporter_node → END

run_workflow（CLI 入口）一次性 ainvoke，并设 recursion_limit=50：

1
result = await workflow.ainvoke(initial_state, config={"recursion_limit": 50})

六、run_guide_chat vs 完整 workflow

这是 iCan 里最容易混淆的两条路径：

路径 入口 是否走 route_after_guide 典型调用方
单轮对话 run_guide_chat 否 api/routes/chat.py 的 /chat、/sessions
完整图 run_workflow → create_workflow 是 cli.py

run_guide_chat 内部调用 agents/guide.py 的 run_guide_single_turn——直接调 LLM，不跑 Guide 子图循环。用户每发一条 HTTP 消息才推进一轮；信息充分后由 API 层 asyncio.create_task 调 run_analysis_pipeline，不再经过顶层 LangGraph。

1
2
3
4
async def run_guide_chat(conversation_history: list, user_message: str) -> dict:
    from ican.agents.guide import run_guide_single_turn
    result = await run_guide_single_turn(conversation_history, user_message)
    # 返回 reply、is_info_sufficient、conversation_history

设计原因：Web 场景需要「等用户输入再往下走」，不能把多轮对话塞进一次 ainvoke 里空转；route_after_guide 的 user_msg_count >= 3 强制退出，主要是给 CLI 一次性跑完整图 时防死循环用的。

七、三重保险防止无限循环

保险层 位置 机制
第 1 层 agents/guide.py should_continue loop_count >= 8 强制 handoff
第 2 层 workflow.py route_after_guide user_msg_count >= 3 强制进 resume_parser_node
第 3 层 LangGraph 框架 外层 recursion_limit=50，子图 recursion_limit=15

三层阈值含义不同：内层限制 子图消息条数，外层限制 顶层 conversation_history 里 user 角色条数，框架限制 图步数总和。

八、踩坑与边界

  1. 注释写「最多 2 次」，代码是 3 轮 user 消息
    route_after_guide 和 should_continue 的 docstring 都写「最多循环 2 次」，实际判断是 user_msg_count >= 3 / loop_count >= 8。写文档和改阈值时要对源码，不要抄注释。

  2. loop_count 数的是 messages 非空条数，不是 user 轮次
    should_continue 用 GuideState.messages（Annotated reducer 累积），和顶层 conversation_history 的计数方式不一致。调循环上限时要分别验证两层 state。

  3. API 路径没有外层 route_after_guide
    若只在浏览器里测 /chat，看不到 user_msg_count >= 3 的效果；要在 CLI 跑 run_workflow 或单测顶层图才能验证强制退出。

  4. needs_more_info 默认 True
    core/state.py 初始 workflow state 里 needs_more_info 为 False，但 route_after_guide 用 state.get("needs_more_info", True)——字段缺失时会偏向继续 guide，设计上是保守策略。

九、小结

  • 条件路由用 add_conditional_edges(source, router_fn, edge_map)，router 只读 state、返回字符串。
  • Guide 内层在 agents/guide.py 用 should_continue 控制 dig_deeper ↔ END；外层在 workflow.py 用 route_after_guide 控制 guide 环 ↔ 分析链。
  • 生产 API 用 run_guide_chat 单轮 + 用户驱动循环；CLI 用 run_workflow 跑完整图，才触发 route_after_guide 的轮次上限。
  • 循环必须有独立计数 + 框架 recursion_limit，避免无用户输入时 LangGraph 空转。
  • 下一篇（第 6 篇）讲双层 StateGraph 嵌套如何把 Guide 子图和顶层编排解耦。

附录：关键源码（逐行注释）

以下代码摘自 iCan 实现，每行上方均有中文注释，不公开仓库也可跟读。
生成命令：python3 bin/build-ican-annotated-snippets.py

内层 should_continue

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# ========== 内层 should_continue ==========
# 源文件: agents/guide.py  行 336-372

# L336: 同步函数 should_continue：路由决策或工厂方法
def should_continue(state: GuideState) -> str:
# L338: 【文档】条件路由函数：判断是继续对话还是交给下一个 Agent。
# L340: 【文档】功能描述：
# L341: 【文档】根据信息充分性检查的结果，决定工作流的走向：
# L342: 【文档】- 如果信息充分（is_info_sufficient=True），流程结束，交给下一个 Agent
# L343: 【文档】- 如果信息不足（is_info_sufficient=False），返回 dig_deeper 继续对话
# L344: 【文档】- 最多循环2次，避免在没有用户实时交互时无限循环
# L346: 【文档】入参说明：
# L347: 【文档】state (GuideState): 对话引导状态对象，需包含 is_info_sufficient 字段。
# L349: 【文档】出参说明：
# L350: 【文档】str: 节点名称字符串，"dig_deeper" 表示继续对话，"handoff" 表示交给下一个 Agent。
# （L337-351 为函数/模块文档字符串，已转为注释便于阅读）
# L352: 开始 try 块，后续 except 负责兜底
    try:
# L353: 记录日志，便于线上排查节点入参/出参
        logger.info("[should_continue] 开始执行，入参: state=%s", {k: str(v)[:100] for k, v in state.items()})
# L354: Guide 判定用户信息是否足够进入分析阶段
        is_sufficient = state.get("is_info_sufficient", False)

# L356: 赋值：更新局部变量或 state 字段
        messages_list = state.get("messages", [])
# L357: 赋值：更新局部变量或 state 字段
        loop_count = len([m for m in messages_list if m])

# L359: 条件分支
        if is_sufficient:
# L360: 记录日志，便于线上排查节点入参/出参
            logger.info("[should_continue] 信息充分，路由到 handoff（交给下一个 Agent）")
# L361: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
            return "handoff"

# L363: 条件分支
        if loop_count >= 8:
# L364: 记录日志，便于线上排查节点入参/出参
            logger.info("[should_continue] 已达到最大循环次数(%d)，强制路由到 handoff", loop_count)
# L365: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
            return "handoff"

# L367: 记录日志，便于线上排查节点入参/出参
        logger.info("[should_continue] 信息不足，路由到 dig_deeper（继续深度挖掘），当前循环=%d", loop_count)
# L368: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
        return "dig_deeper"

# L370: 捕获异常，避免整图/整请求崩溃
    except Exception as e:
# L371: 记录日志，便于线上排查节点入参/出参
        logger.error("[should_continue] 条件路由执行异常: %s", e, exc_info=True)
# L372: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
        return "dig_deeper"

外层 route_after_guide

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# ========== 外层 route_after_guide ==========
# 源文件: workflow.py  行 393-429

# L393: 同步函数 route_after_guide：路由决策或工厂方法
def route_after_guide(state: iCanWorkflowState) -> str:
# L395: 【文档】GuideAgent 后的路由决策：信息充分则继续，否则循环（最多2次）
# L397: 【文档】功能描述：
# L398: 【文档】根据对话引导阶段的信息充分性判断结果，决定工作流的走向：
# L399: 【文档】- 如果 needs_more_info 为 True（信息不足），返回 guide_node 继续对话
# L400: 【文档】- 如果 needs_more_info 为 False（信息充分），返回 resume_parser_node 进入下一阶段
# L401: 【文档】- 最多循环2次，避免在没有用户实时交互时无限循环
# L403: 【文档】入参：
# L404: 【文档】state (iCanWorkflowState): 顶层工作流状态，需包含 needs_more_info 字段
# L406: 【文档】出参：
# L407: 【文档】str: 下一个节点名称，"guide_node" 或 "resume_parser_node"
# （L394-408 为函数/模块文档字符串，已转为注释便于阅读）
# L409: 开始 try 块，后续 except 负责兜底
    try:
# L410: 是否继续 Guide 循环；False 表示可以进 resume_parser
        logger.info("[route_after_guide] 开始执行，入参: needs_more_info=%s", state.get("needs_more_info"))

# L412: 是否继续 Guide 循环；False 表示可以进 resume_parser
        needs_more_info = state.get("needs_more_info", True)
# L413: 多轮对话列表，元素为 {role, content}
        conversation_history = state.get("conversation_history", [])
# L414: 多轮对话列表，元素为 {role, content}
        user_msg_count = len([m for m in conversation_history if m.get("role") == "user"])

# L416: 是否继续 Guide 循环；False 表示可以进 resume_parser
        if not needs_more_info:
# L417: 记录日志，便于线上排查节点入参/出参
            logger.info("[route_after_guide] 路由决策: 信息充分，进入 resume_parser_node")
# L418: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
            return "resume_parser_node"

# L420: 条件分支
        if user_msg_count >= 3:
# L421: 记录日志，便于线上排查节点入参/出参
            logger.info("[route_after_guide] 路由决策: 已有%d轮用户消息，强制进入 resume_parser_node", user_msg_count)
# L422: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
            return "resume_parser_node"

# L424: 记录日志，便于线上排查节点入参/出参
        logger.info("[route_after_guide] 路由决策: 信息不足，返回 guide_node 继续对话")
# L425: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
        return "guide_node"

# L427: 捕获异常，避免整图/整请求崩溃
    except Exception as e:
# L428: 记录日志，便于线上排查节点入参/出参
        logger.error("[route_after_guide] 路由决策执行异常: %s", e, exc_info=True)
# L429: 返回本节点要合并进 state 的字段（LangGraph 会 merge）
        return "resume_parser_node"

条件边注册

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# ========== 条件边注册 ==========
# 源文件: workflow.py  行 432-474

# L432: 同步函数 create_workflow：路由决策或工厂方法
def create_workflow() -> StateGraph:
# L434: 【文档】创建顶层工作流 StateGraph
# L436: 【文档】功能描述：
# L437: 【文档】构建 iCan 系统的顶层 LangGraph 工作流图，将 5 个 Agent 节点串联起来，
# L438: 【文档】并通过条件路由实现 GuideAgent 的循环对话机制。
# L440: 【文档】入参：无
# L442: 【文档】出参：
# L443: 【文档】StateGraph: 编译后的 LangGraph StateGraph 实例
# L445: 【文档】工作流：
# L446: 【文档】start -> guide_node -> route_after_guide
# L447: 【文档】-> (信息不足) -> guide_node [循环]
# L448: 【文档】-> (信息充分) -> resume_parser_node -> profile_analyzer_node
# L449: 【文档】-> career_matcher_node -> reporter_node -> END
# （L433-450 为函数/模块文档字符串，已转为注释便于阅读）
# L451: 开始 try 块，后续 except 负责兜底
    try:
# L452: 记录日志，便于线上排查节点入参/出参
        logger.info("[create_workflow] 开始创建顶层工作流 StateGraph")

# L454: 创建 LangGraph 状态图，括号内 TypedDict 定义各节点共享/传递的字段
        graph = StateGraph(iCanWorkflowState)

# L456: 添加节点
# L457: 注册图节点「guide_node」，值为 async 节点函数
        graph.add_node("guide_node", guide_node)
# L458: 注册图节点「resume_parser_node」，值为 async 节点函数
        graph.add_node("resume_parser_node", resume_parser_node)
# L459: 注册图节点「profile_analyzer_node」，值为 async 节点函数
        graph.add_node("profile_analyzer_node", profile_analyzer_node)
# L460: 注册图节点「career_matcher_node」，值为 async 节点函数
        graph.add_node("career_matcher_node", career_matcher_node)
# L461: 注册图节点「reporter_node」，值为 async 节点函数
        graph.add_node("reporter_node", reporter_node)

# L463: 设置入口节点
# L464: 设置图入口：ainvoke 时第一个执行的节点
        graph.set_entry_point("guide_node")

# L466: 定义条件边：guide_node 后根据信息充分性路由
# L467: 添加条件边：由路由函数返回值决定下一节点名
        graph.add_conditional_edges(
# L468: 执行该语句（细节见上文业务描述）
            "guide_node",
# L469: 执行该语句（细节见上文业务描述）
            route_after_guide,
# L470: 执行该语句（细节见上文业务描述）
            {
# L471: 执行该语句（细节见上文业务描述）
                "guide_node": "guide_node",
# L472: 执行该语句（细节见上文业务描述）
                "resume_parser_node": "resume_parser_node",
# L473: 执行该语句（细节见上文业务描述）
            },
# L474: 执行该语句（细节见上文业务描述）
        )

系列导航

篇 主题
1 系统全景
2 五 Agent 协作
3 霍兰德 RIASEC
4–7 状态 · 5 路由（本篇） · 嵌套 · 容错
8–11 LLM 层 · SSE/WS · DB 迁移 · PDF
12–14 JSON Prompt · RIASEC Prompt · Guide Prompt
15–17 Docker · 中间件 · 配置

← 返回 iCan 专题