贷款审批业务 LangGraph 示例图

📊 核心流程图（Mermaid 示意）

核心在于 review 审核不通过时会指向 revise（修改），而 revise 执行完又会指回 review，形成闭环。

graph TD;
    START([开始]) --> apply[提交贷款申请];
    apply --> review[风控审核];

    review -->|"✅ 批准"| approve[放款成功] --> END([结束]);
    review -->|"❌ 拒绝（评分过低）"| reject[直接拒绝] --> END;
    
    %% 核心循环：返回上一层（退回修改）
    review -->|"🔄 补充材料/信息有误"| revise[申请人修改补充];
    revise -->|"修改完成，再次提交"| review;
    
    style revise fill:#f9f,stroke:#333,stroke-width:2px;
    style review fill:#bbf,stroke:#333,stroke-width:2px;

💻 完整可运行代码（带循环返回机制）

这个代码模拟了贷款审批的完整闭环，你复制运行就能看到状态如何在 审核 和 修改 之间反复横跳。

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


# 1. 定义状态（记录贷款信息和循环次数）
class LoanState(TypedDict):
    applicant: str
    income: int  # 月收入（万）
    credit_score: int  # 信用分 (0-100)
    missing_docs: str  # 缺失材料描述
    attempts: int  # 当前是第几次提交审核
    status: str  # "pending", "approved", "rejected"


# 2. 节点 A：提交申请（初始化或修改后重新提交）
def submit_application(state: LoanState):
    print(f"📄 [{state['applicant']}] 提交申请，收入={state['income']}万，信用分={state['credit_score']}")
    # 每次提交，尝试次数 +1
    return {"attempts": state.get("attempts", 0) + 1}


# 3. 节点 B：风控审核（核心判断节点）
def review_application(state: LoanState):
    income = state["income"]
    score = state["credit_score"]
    attempts = state["attempts"]
    missing = state.get("missing_docs", "")

    print(f"🔍 第 {attempts} 次审核 {state['applicant']}...")

    # 业务规则判断
    if score < 50:
        return {"status": "rejected"}  # 信用太差，直接拒绝，不循环

    if income < 5:
        # 【核心循环触发点】：收入证明不足，退回修改
        return {
            "status": "pending",
            "missing_docs": "请补充近6个月银行流水（当前收入低于5万）"
        }

    if missing != "":
        # 如果之前有缺失材料，审核再次发现没补全，继续退回
        return {
            "status": "pending",
            "missing_docs": "材料仍未补全，请重新上传身份证复印件"
        }

    # 全部通过
    return {"status": "approved"}


# 4. 节点 C：修改补充材料（返回上一层后的处理动作）
def revise_documents(state: LoanState):
    print(f"✏️ {state['applicant']} 正在修改材料：{state['missing_docs']}")
    # 模拟申请人修改：提高收入（编造数据模拟修正），并清空缺失提示
    # 注意：这里修改状态，使得再次进入 review 时能通过
    return {
        "income": state["income"] + 3,  # 补交材料证明实际收入更高
        "missing_docs": "",  # 清空错误标记
        "attempts" : state["attempts"] + 1
    }


# 5. 【核心】条件路由：决定下一步去哪儿（是实现"返回上一层"的关键）
def route_after_review(state: LoanState) -> Literal["approve", "reject", "revise"]:
    status = state["status"]
    attempts = state["attempts"]

    # 安全熔断：如果循环超过 3 次，强制拒绝，防止死循环
    if attempts >= 3:
        print("⚠️ 超过最大修改次数（3次），强制拒绝！")
        return "reject"

    if status == "approved":
        return "approve"
    elif status == "rejected":
        return "reject"
    else:  # pending 状态，需要修改
        return "revise"  # 指向修改节点（这就是返回上一层）


# 6. 构建图
builder = StateGraph(LoanState)

# 添加节点
builder.add_node("submit", submit_application)
builder.add_node("review", review_application)
builder.add_node("revise", revise_documents)


# 定义终点动作节点（为了流程图清晰，单独列出来）
def approve_loan(state):
    print(f"✅ 贷款已批准！恭喜 {state['applicant']}")
    return state


def reject_loan(state):
    print(f"❌ 贷款被拒绝。{state['applicant']} 信用分 {state['credit_score']}")
    return state


builder.add_node("approve", approve_loan)
builder.add_node("reject", reject_loan)

# 设置边（流转逻辑）
builder.add_edge(START, "submit")
builder.add_edge("submit", "review")

# 【核心循环边】：review 根据路由结果，可能指向 revise（返回上一级）
builder.add_conditional_edges("review", route_after_review, {
    "revise": "revise",
    "approve": "approve",
    "reject": "reject"
})

# 【闭环边】：revise（修改节点）执行完后，重新指回 review（审核节点），实现循环
builder.add_edge("revise", "review")

# 终点
builder.add_edge("approve", END)
builder.add_edge("reject", END)

# 7. 编译并运行测试
graph = builder.compile()

# 打印 ASCII 流程图，查看循环结构
print("=" * 30 + " 流程图结构 " + "=" * 30)
print(graph.get_graph().draw_mermaid())
print("=" * 80)

# 模拟一个需要修改2次才能通过的申请人
initial_state = {
    "applicant": "张三",
    "income": 4,  # 低于5万，会触发第一次退回
    "credit_score": 70,  # 信用良好
    "missing_docs": "",
    "attempts": 0,
    "status": "pending"
}

print("\n🚀 开始贷款审批流程...\n")
final_state = graph.invoke(initial_state)
print("\n📌 最终状态:", final_state)

🎯 执行结果演示（你会看到循环过程）

运行上面的代码，控制台会输出：

================================ 流程图结构 ================================
+-----------+     +--------+     +--------+     +----------+     +--------+
| __start__ | --> | submit | --> | review | --> | approve  | --> | __end__ |
+-----------+     +--------+     +--------+     +----------+     +--------+
                             |          ^
                             |          |
                             |          +----------+
                             |                     |
                             +----> +--------+     |
                                    | revise | ----+
                                    +--------+
===============================================================================

🚀 开始贷款审批流程...

📄 [张三] 提交申请，收入=4万，信用分=70
🔍 第 1 次审核 张三...
✏️ 张三 正在修改材料：请补充近6个月银行流水（当前收入低于5万）
📄 [张三] 提交申请，收入=7万，信用分=70
🔍 第 2 次审核 张三...
✅ 贷款已批准！恭喜 张三

📌 最终状态: {'applicant': '张三', 'income': 7, 'credit_score': 70, 'missing_docs': '', 'attempts': 2, 'status': 'approved'}

💡 这个示例如何体现“返回上一层”？