# State Management
    class State(TypedDict):
        messages: Annotated[List[AnyMessage], add_messages]
​
    # Nodes
    def chat_node(state: State) -> State:
        state["messages"] = chat_llm.invoke({"messages": state["messages"]})
        return state
​
    # Building the graph
    graph_builder = StateGraph(State)
    graph_builder.add_node("chat_node", chat_node)
    graph_builder.add_node("tool_node", ToolNode(tools=tools))
    graph_builder.add_edge(START, "chat_node")
    graph_builder.add_conditional_edges("chat_node", tools_condition, {"tools": "tool_node", "__end__": END})
    graph_builder.add_edge("tool_node", "chat_node")
    graph = graph_builder.compile(checkpointer=MemorySaver())
    return graph

class ToolNode:
    def __init__(self, tools):
        self.tools = tools
    
    def __call__(self, state: State) -> State:
        # 执行工具调用
        tool_results = []
        for tool_call in state["messages"][-1].tool_calls:
            tool = self.tools[tool_call["name"]]
            result = tool.invoke(tool_call["args"])
            tool_results.append(result)
        
        return {"messages": tool_results}

def tools_condition(state: State) -> str:
    """判断是否需要调用工具"""
    # 获取最后一条消息
    last_message = state["messages"][-1]
    
    # 检查是否是工具调用请求
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"  # 需要调用工具
    else:
        return "__end__"  # 结束对话

def tools_condition(
    state: list[AnyMessage] | dict[str, Any] | BaseModel,
    messages_key: str = "messages",
) -> Literal["tools", "__end__"]:
    """Conditional routing function for tool-calling workflows.

    This utility function implements the standard conditional logic for ReAct-style
    agents: if the last `AIMessage` contains tool calls, route to the tool execution
    node; otherwise, end the workflow. This pattern is fundamental to most tool-calling
    agent architectures.

    The function handles multiple state formats commonly used in LangGraph applications,
    making it flexible for different graph designs while maintaining consistent behavior.

    Args:
        state: The current graph state to examine for tool calls. Supported formats:
            - Dictionary containing a messages key (for `StateGraph`)
            - `BaseModel` instance with a messages attribute
        messages_key: The key or attribute name containing the message list in the state.
            This allows customization for graphs using different state schemas.

    Returns:
        Either `'tools'` if tool calls are present in the last `AIMessage`, or `'__end__'`
            to terminate the workflow. These are the standard routing destinations for
            tool-calling conditional edges.

    Raises:
        ValueError: If no messages can be found in the provided state format.

    Example:
        Basic usage in a ReAct agent:

        ```python
        from langgraph.graph import StateGraph
        from langchain.tools import ToolNode
        from langchain.tools.tool_node import tools_condition
        from typing_extensions import TypedDict


        class State(TypedDict):
            messages: list


        graph = StateGraph(State)
        graph.add_node("llm", call_model)
        graph.add_node("tools", ToolNode([my_tool]))
        graph.add_conditional_edges(
            "llm",
            tools_condition,  # Routes to "tools" or "__end__"
            {"tools": "tools", "__end__": "__end__"},
        )
        ```

        Custom messages key:

        ```python
        def custom_condition(state):
            return tools_condition(state, messages_key="chat_history")
        ```

    !!! note
        This function is designed to work seamlessly with `ToolNode` and standard
        LangGraph patterns. It expects the last message to be an `AIMessage` when
        tool calls are present, which is the standard output format for tool-calling
        language models.
    """
    if isinstance(state, list):
        ai_message = state[-1]
    elif (isinstance(state, dict) and (messages := state.get(messages_key, []))) or (
        messages := getattr(state, messages_key, [])
    ):
        ai_message = messages[-1]
    else:
        msg = f"No messages found in input state to tool_edge: {state}"
        raise ValueError(msg)
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return "__end__"

{
    "tools": "tool_node",  # 当条件返回 "tools" 时，跳转到 tool_node
    "__end__": END         # 当条件返回 "__end__" 时，结束工作流
}

def advanced_condition(state: State) -> str:
    last_message = state["messages"][-1]
    
    if "help" in last_message.content:
        return "help_flow"
    elif "purchase" in last_message.content:
        return "checkout_flow"
    elif "cancel" in last_message.content:
        return "cancellation_flow"
    else:
        return "__end__"

graph_builder.add_conditional_edges(
    "chat_node",
    advanced_condition,
    {
        "help_flow": "help_node",
        "checkout_flow": "checkout_node",
        "cancellation_flow": "cancellation_node",
        "__end__": END
    }
)

# 第一层条件分支
graph_builder.add_conditional_edges(
    "initial_node",
    determine_flow_type,
    {"support": "support_flow", "sales": "sales_flow"}
)

# 支持流中的子分支
graph_builder.add_conditional_edges(
    "support_flow",
    support_condition,
    {"technical": "tech_support_node", "billing": "billing_support_node"}
)

# 销售流中的子分支
graph_builder.add_conditional_edges(
    "sales_flow",
    sales_condition,
    {"new": "new_customer_node", "existing": "existing_customer_node"}
)

def safe_condition(state: State) -> str:
    try:
        # 业务逻辑
    except Exception as e:
        # 记录错误
        state["errors"].append(str(e))
        return "error_handling"