1. LangGraph实战：从零构建智能审核系统

第一次接触LangGraph时，我被它处理复杂逻辑的能力惊艳到了。当时我需要给一个社区平台开发内容审核系统，传统if-else代码写了200多行还是漏洞百出。直到把审核流程画成状态图，用节点处理具体规则，用边控制流程跳转，整个系统突然变得清晰可控。下面我就用这个真实案例，带你体验如何用LangGraph构建智能决策工作流。

先看个典型场景：用户发布包含图片的帖子时，系统需要先检查图片合规性，再分析文本内容，根据两者的综合结果决定"直接发布"、"进入人工复核"或"拒绝"。用传统方法实现这种多条件分支，代码会变成难以维护的"面条式"逻辑。而用LangGraph只需要定义三个核心组件：

• 图（Graph）：整个审核流程的容器，维护着当前审核状态（如原始内容、中间结果、最终决策）
• 节点（Node）：每个审核步骤的具体执行单元（如图片识别、敏感词检测）
• 边（Edge）：决定审核流程走向的逻辑（如"当图片违规时跳转到拒绝节点"）

from langgraph.graph import StateGraph

# 定义初始状态结构
def init_state():
    return {
        "content": "",
        "image_check": None,
        "text_check": None,
        "final_decision": None
    }

# 构建图框架
builder = StateGraph(init_state)

2. 设计审核流程的节点与边

2.1 实现图片审核节点

图片审核节点需要调用视觉AI服务，我使用的是开源的CLIP模型。这个节点的特殊之处在于：它需要处理可能出现的网络超时异常，并返回标准化的检查结果。

def image_check_node(state):
    try:
        # 模拟调用图片审核API
        from clip_client import Client
        client = Client()
        result = client.analyze(state["content"]["image"])
        
        # 标准化输出
        return {
            "image_check": {
                "is_violent": result.violence_score > 0.7,
                "is_explicit": result.explicit_score > 0.8,
                "error": None
            }
        }
    except Exception as e:
        return {
            "image_check": {
                "error": str(e),
                "is_violent": False,
                "is_explicit": False
            }
        }

# 将节点添加到图中
builder.add_node("image_check", image_check_node)

2.2 构建条件跳转逻辑

审核系统的核心智能体现在边的条件判断上。我们定义三种边类型：

1. 强制拒绝边：当图片包含暴力或色情内容时直接终止流程
2. 文本复核边：图片正常时进入文本审核
3. 异常处理边：当服务调用失败时进入人工复核

def should_reject(state):
    checks = state["image_check"]
    if checks["error"]:
        return "human_review"  # 服务异常转人工
    elif checks["is_violent"] or checks["is_explicit"]:
        return "reject"  # 违规内容直接拒绝
    else:
        return "text_check"  # 进入文本审核

# 添加条件边
builder.add_conditional_edges(
    "image_check",
    should_reject,
    {
        "reject": END,
        "human_review": "human_review_node",
        "text_check": "text_check_node"
    }
)

3. 处理复杂文本审核场景

3.1 多维度文本分析节点

文本审核需要同时检查：

• 敏感词（使用正则表达式匹配）
• 情感倾向（调用情感分析API）
• 话题相关性（用嵌入向量计算相似度）

def text_check_node(state):
    content = state["content"]["text"]
    
    # 敏感词检测
    banned_words = ["暴力", "违禁品"]
    has_banned = any(word in content for word in banned_words)
    
    # 情感分析（模拟）
    sentiment = "positive" if len(content) < 50 else "neutral"
    
    # 话题相关性（实际项目会用BERT等模型）
    is_relevant = "技术" in content or "编程" in content
    
    return {
        "text_check": {
            "has_banned": has_banned,
            "sentiment": sentiment,
            "is_relevant": is_relevant
        }
    }

builder.add_node("text_check", text_check_node)

3.2 动态决策边设计

文本审核后的决策更复杂，我们引入权重计算：

• 含敏感词直接拒绝
• 负面情感且无关话题进入人工复核
• 其他情况自动通过

def text_decision(state):
    checks = state["text_check"]
    
    if checks["has_banned"]:
        return "reject"
        
    sentiment_score = 1 if checks["sentiment"] == "positive" else 0
    relevance_score = 2 if checks["is_relevant"] else 0
    total = sentiment_score + relevance_score
    
    if total < 1:
        return "human_review"
    else:
        return "approve"

builder.add_conditional_edges(
    "text_check",
    text_decision,
    {
        "reject": END,
        "human_review": "human_review_node",
        "approve": "publish_node"
    }
)

4. 实现自我修正的工作流

4.1 人工复核的闭环设计

当系统不确定时，会进入人工复核节点。关键点在于：人工决策结果应该反馈给系统，用于优化未来的自动审核。

def human_review_node(state):
    # 模拟人工审核界面
    print(f"待审核内容：{state['content']}")
    decision = input("输入决策（approve/reject）：")
    
    # 记录人工决策用于后续模型训练
    log_decision(state, decision)
    
    return {"final_decision": decision}

builder.add_node("human_review", human_review_node)

# 人工复核后统一结束流程
builder.add_edge("human_review", END)

4.2 发布节点的后处理

通过审核的内容在发布前还需要：

1. 内容格式化（如Markdown转换）
2. 打上审核通过的标签
3. 触发通知流程

def publish_node(state):
    formatted = format_content(state["content"])
    tagged = add_audit_tag(formatted)
    notify_user(tagged)
    return {"final_decision": "published"}

builder.add_node("publish", publish_node)
builder.add_edge("publish", END)

4.3 编译并测试完整工作流

最后编译整个图并测试不同场景：

# 编译图
audit_graph = builder.compile()

# 测试违规内容
bad_post = {"text": "出售违禁品", "image": "暴力图片"}
print(audit_graph.invoke({"content": bad_post}))
# 预期输出：final_decision为reject

# 测试边缘案例
ambiguous_post = {"text": "这个产品很差", "image": "正常图片"}
print(audit_graph.invoke({"content": ambiguous_post}))
# 预期进入human_review节点

实际项目中，我们会用单元测试覆盖所有条件分支。LangGraph的可视化工具还能生成流程图，非常便于团队协作和流程优化。