LangGraph条件工作流：无需提示词即可实现分支处理

本指南旨在为“LangGraph中的条件工作流：智能代理的If-Else结构”重建可操作的实现路径。重点介绍那些可以直接放入代码库而无需猜测其用途的契约、检查项及代码。在开始修改代码之前，应先明确输入参数、各步骤的负责人以及终止标准。操作人员应当能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。同时需将正常流程与异常恢复路径一并记录下来。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续需要补充的内容。

什么是条件工作流？

那么，条件化工作流究竟有什么用呢？在修改代码之前，应先明确输入参数、该步骤的负责人以及终止条件。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。 相比庞大的脚本，更应采用小型且可测试的单元。当某个步骤出现故障时，故障原因应当指向单一的责任主体，而非错综复杂的流程。 严格限制工具的架构。过长的自由文本参数容易引发注入攻击，还会增加审计成本。

工作流 #1：求解二次方程

对于工作流#1：求解二次方程，在修改代码之前需明确输入参数、各步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 严格限制工具的结构规范。过宽的自由文本参数容易引发注入攻击，且会增加审计成本。

工作流结构

在修改代码之前，需先明确工作流结构中的输入参数、各步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测其中的隐藏状态。应在功能结果旁记录执行时间和成本，提前明确这些信息可避免在流程从演示环境转向共享环境时出现意外费用。需严格限制工具的架构，过宽的自由文本参数容易引发注入攻击，并增加审计成本。在修改代码之前，需先明确工作流结构中的输入参数、各步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测其中的隐藏状态。需同时记录正常流程和异常恢复流程，重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。

构建状态

在构建状态时，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任主体，而非复杂的流程链。 在耗时的模型调用之后设置检查点，这样重试时就不会重复处理相同的工作。

class QuadState(TypedDict):
    a: float
    b: float
    c: float
    equation: str
    discriminant: float
    result: str

节点

对于各个节点，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 在执行耗时的模型调用后设置检查点，以避免重复处理相同任务时再次产生费用。

def show_equation(state: QuadState):
    equation = f"{state['a']}x² + {state['b']}x + {state['c']}"
    return {"equation": equation}

def calculate_discriminant(state: QuadState):
    discriminant = (state['b'] ** 2) - (4 * state['a'] * state['c'])
    return {"discriminant": discriminant}

def real_roots(state: QuadState):
    r1 = (-state['b'] + state['discriminant'] ** 0.5) / (2 * state['a'])
    r2 = (-state['b'] - state['discriminant'] ** 0.5) / (2 * state['a'])
    result = f"The roots are {r1} and {r2}"
    return {"result": result}

def repeated_root(state: QuadState):
    root = -state['b'] / (2 * state['a'])
    result = f"Only repeating root is {root}"
    return {"result": result}

def no_real_roots(state: QuadState):
    return {"result": "No real roots"}

路由功能——这是核心部分

对于路由功能——这是最关键的部分，在修改代码之前必须明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间和成本。提前明确这些信息可以避免在流程从演示环境切换到共享环境时出现意外费用。 在耗时的模型调用之后设置检查点，这样重新尝试时就不会重复计算相同的工作量。 对于路由功能——这是最关键的部分，在修改代码之前必须明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的功能。

def check_condition(state: QuadState):
    if state['discriminant'] > 0:
        return "real_roots"
    elif state['discriminant'] == 0:
        return "repeated_root"
    else:
        return "no_real_roots"

将各部分整合起来

在将各部分连接起来时，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一的责任主体，而非错综复杂的流程。 将规划与工具执行分开。规划者负责提出方案；执行者负责进行操作；验证者则根据目标检查结果。

graph = StateGraph(QuadState)

graph.add_node("show_equation", show_equation)
graph.add_node("calculate_discriminant", calculate_discriminant)
graph.add_node("real_roots", real_roots)
graph.add_node("repeated_root", repeated_root)
graph.add_node("no_real_roots", no_real_roots)

graph.add_edge(START, "show_equation")
graph.add_edge("show_equation", "calculate_discriminant")

# This is the magic line
graph.add_conditional_edges("calculate_discriminant", check_condition)

graph.add_edge("real_roots", END)
graph.add_edge("repeated_root", END)
graph.add_edge("no_real_roots", END)

workflow = graph.compile()

工作流#2：客户评价处理系统（基于大语言模型）

对于工作流#2：客户评价处理模块（基于大语言模型），在修改代码之前需明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 将规划与工具执行分开。规划者负责提出方案，执行者负责进行操作，验证者则根据目标检查结果。

场景描述

针对该场景，在修改代码之前需明确输入参数、各步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间与成本。提前明确这些信息，可避免在流程从演示环境转向共享环境时出现意外费用。应将规划与工具执行分开：规划者负责提出方案，执行者负责实施变更，验证者则根据目标检查最终结果。针对该场景，在修改代码之前需明确输入参数、各步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。

结构化输出模式

对于结构化输出方案，在修改代码之前需明确输入内容、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 严格限制工具方案的结构。过大的自由文本参数容易引发注入攻击，且会增加审计成本。

from pydantic import BaseModel
from typing import Literal

class SentimentSchema(BaseModel):
    sentiment: Literal["positive", "negative"]

class DiagnosisSchema(BaseModel):
    issue_type: Literal["UI", "performance", "bug", "support", "other"]
    tone: Literal["frustrated", "angry", "disappointed", "neutral"]
    urgency: Literal["low", "medium", "high"]

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
structured_model_sentiment = model.with_structured_output(SentimentSchema)
structured_model_diagnosis = model.with_structured_output(DiagnosisSchema)

状态

对于国家系统而言，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 严格限制工具的结构规范。过宽的自由文本参数容易引发注入攻击，还会增加审计成本。

class ReviewState(TypedDict):
    review: str
    sentiment: Literal["positive", "negative"]
    diagnosis: dict
    response: str

节点

对于The Nodes，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间和成本。提前明确相关内容可避免在流程从演示环境转向共享环境时出现意外费用。需严格限制工具的架构，过宽的自由文本参数容易引发注入攻击，并增加审计成本。对于The Nodes，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。

def find_sentiment(state: ReviewState):
    prompt = f"For the following review, find out the sentiment:\n{state['review']}"
    result = structured_model_sentiment.invoke(prompt)
    return {"sentiment": result.sentiment}

def positive_response(state: ReviewState):
    prompt = f"""Write a warm thank you message in response to this review:
    {state['review']}
    Also kindly ask the user to leave feedback on our website."""

    result = model.invoke(prompt)
    return {"response": result.content}

def run_diagnosis(state: ReviewState):
    prompt = f"""Diagnose this negative review:
    {state['review']}
    Return issue type, tone, and urgency."""

    result = structured_model_diagnosis.invoke(prompt)
    return {"diagnosis": result.model_dump()}

def negative_response(state: ReviewState):
    diagnosis = state['diagnosis']
    prompt = f"""You are a support assistant. The user had a {diagnosis['issue_type']} issue,
    sounded {diagnosis['tone']}, and marked urgency as {diagnosis['urgency']}.
    Write an empathetic, helpful resolution message."""

    result = model.invoke(prompt)
    return {"response": result.content}

def check_sentiment(state: ReviewState):
    if state['sentiment'] == "positive":
        return "positive_response"
    else:
        return "run_diagnosis"

图结构

对于 The Graph，修改代码之前需先定义输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相比庞大的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个特定职责，而非整个复杂的流程。 在执行耗时的模型调用后设置检查点，这样重新尝试时就不会重复处理相同的工作。

graph = StateGraph(ReviewState)

graph.add_node("find_sentiment", find_sentiment)
graph.add_node("positive_response", positive_response)
graph.add_node("run_diagnosis", run_diagnosis)
graph.add_node("negative_response", negative_response)

graph.add_edge(START, "find_sentiment")
graph.add_conditional_edges("find_sentiment", check_sentiment)
graph.add_edge("positive_response", END)
graph.add_edge("run_diagnosis", "negative_response")
graph.add_edge("negative_response", END)

workflow = graph.compile()

测试它

为进行测试，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 在耗时的模型调用之后设置检查点，以避免重复执行相同任务时再次产生费用。

核心思维模型

对于核心思维模型，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间和成本。提前明确这些信息可避免在流程从演示环境转向共享环境时出现意外费用。在高成本模型调用之后设置检查点，这样重试时就不会重复计算相同的工作量。对于核心思维模型，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。

下一步计划

在下一步操作中，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。需将规划与工具执行分开：规划者负责提出方案，执行者负责进行修改，验证者则根据目标检查最终结果。

超越博客的结构化学习

需将规划与工具执行分开：规划者负责提出方案，执行者负责进行修改，验证者则根据目标检查最终结果。

课程

需将规划与工具执行分开：规划者负责提出方案，执行者负责进行修改，验证者则根据目标检查最终结果。

操作检查清单

在昂贵的模型调用之后设置检查点，这样重试时就不会重复计算相同的工作。

保持图结构简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会导致中断后无法继续执行。

编写简短的操作手册：如何轮换密钥、如何清空队列、如何回滚最近的更改。