拓扑重构

这个问题问得非常专业，直击图编排（Graph Orchestration）的核心优势——拓扑重构（Topology Refactoring）。

下面我直接给你写可运行的代码，展示如何在不改动 generate 函数内部一行代码的情况下，插入“敏感词过滤”节点。

🛠️ 插入“敏感词过滤”节点的完整写法

我们遵循一个原则：所有流向 generate 的路径，都必须先经过 sensitive_filter。

第一步：定义过滤节点函数

（你可以把它放在 node_generate 的上面或下面，无所谓）

def node_sensitive_filter(state: WorkflowState):
    """节点 3.5: 敏感词过滤（纯逻辑，不依赖 generate）"""
    print("🔞 [步骤 3.5/5] 正在执行敏感词过滤...")

    # 模拟过滤逻辑：检查用户问题或即将生成的上下文
    # 假设发现敏感词，我们可以在 state 里打个标记，或者直接拦截
    if "暴力" in state["user_query"] or "色情" in state["user_query"]:
        print("   ⚠️ 检测到敏感词！已拦截。")
        # 这里可以直接修改 state，让后续的 generate 生成固定回复
        # 但我们为了演示“不改 generate”，选择在 state 里加一个 flag
        return {"skip_normal_answer": True, "final_answer": "抱歉，检测到敏感内容，无法回答。"}
    else:
        print("   ✅ 未检测到敏感词，放行。")
        return {"skip_normal_answer": False}

第二步：将新节点加入 Graph 构建器

# ...（之前的 builder 定义代码）...

# 添加新节点
builder.add_node("sensitive_filter", node_sensitive_filter)

第三步：【核心】修改路由映射（不改 generate 代码）

这是关键。我们不需要动 node_generate 函数里的任何逻辑，只需要修改图中的边（Edges），让所有去往 generate 的人，先拐到 sensitive_filter 那里报到。

修改前（你现在的代码）：

builder.add_conditional_edges(
    "router",
    route_after_router,
    {
        "query_database": "query_database",
        "generate": "generate"   # 直接去 generate
    }
)
builder.add_edge("query_database", "generate")  # 直接去 generate
builder.add_edge("generate", "push_wechat")

修改后（插入过滤器）：

# 1. 修改条件路由：如果 router 决定去 generate，现在先拐去 sensitive_filter
builder.add_conditional_edges(
    "router",
    route_after_router,
    {
        "query_database": "query_database",
        "generate": "sensitive_filter"   # 改成先去过滤器！
    }
)

# 2. 修改静态边：查完数据库也不直接生成，先拐去 sensitive_filter
builder.add_edge("query_database", "sensitive_filter")  # 改成先去过滤器！

# 3. 新增边：过滤器执行完，再去生成（此时 generate 完全不知情）
builder.add_edge("sensitive_filter", "generate")

# 4. 这行完全不用动
builder.add_edge("generate", "push_wechat")

⚙️ 新流程的两种路径执行情况

• 路径 A（不走数据库）：router → sensitive_filter → generate → push_wechat
• 路径 B（走数据库）：router → query_database → sensitive_filter → generate → push_wechat

无论走哪条路，generate 都只接收来自 sensitive_filter 的输入。它完全不知道原本的上游是 router 还是 query_database。

🔍 如果过滤节点想“直接拦截”，不让 generate 执行怎么办？

这就涉及到 LangGraph 的条件边（Conditional Edge）升级用法了。如果你希望检测到敏感词后，跳过 generate，直接去推送微信，你可以这样写：

# 改造 router 后的路由决策
def route_after_filter(state: WorkflowState) -> Literal["generate", "push_wechat"]:
    if state.get("skip_normal_answer", False):
        print("   🚫 因敏感词拦截，跳过生成环节，直接推送。")
        return "push_wechat"   # 直接跳到最后一步
    else:
        return "generate"

# 取消原来的 add_edge("sensitive_filter", "generate")
# 改为条件边：
builder.add_conditional_edges(
    "sensitive_filter",
    route_after_filter,
    {
        "generate": "generate",
        "push_wechat": "push_wechat"
    }
)

此时，generate 函数依然一行不改。

💡 为什么这比传统 if-else 高级？

如果用传统的 Python 代码写，你要么在 generate 函数头部加 if，要么在调用 generate 之前加 if，这违背了开闭原则（对扩展开放，对修改封闭）。

而在图（Graph）中：

1. 新增节点（sensitive_filter）是扩展。
2. 修改边（edge 和 conditional_edge）是重新布线。
3. 原有的 generate 节点被当作黑盒复用，哪怕它是由另一个团队维护的，你也不需要它的源码。

这就是 “图即代码（Graph as Code）” 的魅力——业务逻辑的解耦程度，取决于节点之间边的灵活性，而非节点内部的代码改动。 继续保持这种拆解架构的思维，你已经摸到了高级工程师的门槛！🚀

本作品采用《CC 协议》，转载必须注明作者和本文链接