LangGraph Node节点：让AI”动起来”

想象一下，你走进一家热闹的餐厅。厨房里，厨师正在翻炒锅中的食材；吧台前，调酒师熟练地摇晃着雪克杯；收银台后，收银员微笑着为客人结账。每个人都有自己的职责，每个人都在特定的时刻执行特定的任务，而最终，一道道美味佳肴被端上餐桌。

这就是**节点（Node）**的魅力——将复杂的流程拆解成一个个独立、可复用的单元。

在LangGraph的世界里，Node就是构建AI工作流的基本砖块。如果说LangGraph是一座宏伟的建筑，那么Node就是构成这座建筑的每一块砖石。理解Node，你就掌握了让AI真正”动起来”的钥匙。

一、什么是Node？——餐厅里的分工艺术

1.1 从餐厅说起

让我们继续用餐厅来比喻。一家餐厅要正常运营，需要哪些角色？

• 服务员：接待客人，记录点单，将需求传递给厨房
• 厨师：根据订单烹饪食物，把控口味和品质
• 收银员：计算账单，处理付款，开具发票
• 保洁员：打扫餐桌，保持环境整洁

每个角色都是一个独立的”节点”，他们各自有明确的职责，通过特定的流程（服务员下单→厨师做菜→服务员上菜→收银员结账）协同工作，最终为客人提供完整的用餐体验。

1.2 AI世界的Node

在LangGraph中，Node的作用与餐厅中的角色异曲同工。每个Node都是一个执行特定任务的函数或方法，它们接收输入、处理数据、产生输出，并通过状态（State）与其他Node交换信息。

举个例子，一个最简单的AI对话系统可能包含：

• 理解节点：分析用户的输入，提取关键信息
• 思考节点：根据提取的信息进行推理和决策
• 回复节点：生成自然语言的回应

这三个Node串联起来，就构成了一个完整的对话流程。

1.3 为什么需要Node？

你可能会问：为什么不把所有逻辑写在一个大函数里呢？

好问题！想象一下，如果餐厅只有一个员工，他既要接待客人，又要炒菜，还要收银和打扫——效率会极其低下，而且一旦他请假，整个餐厅就得关门。

使用Node的好处：

1. 职责单一：每个Node只做一件事，代码更易理解和维护
2. 灵活组合：像搭积木一样，不同的Node可以组合出不同的工作流
3. 易于调试：当出现问题时，可以快速定位到具体的Node
4. 便于复用：写好的Node可以在多个项目中重复使用
5. 可视化：LangGraph可以将Node和它们之间的连接绘制成图，一目了然

二、LangGraph中的三种Node类型

LangGraph主要支持三种类型的Node，它们各有特色，适用于不同的场景。

2.1 函数节点（Function Node）——万能的基础砖块

函数节点是最基础、最灵活的Node类型。它本质上就是一个Python函数，你可以在里面写任何逻辑。

适用场景：

• 数据预处理（清洗、格式化）
• 条件判断（根据某些条件决定下一步做什么）
• 状态更新（修改共享状态中的数据）
• 调用外部API
• 任何自定义逻辑

代码示例：

from langgraph.graph import StateGraph
from typing import TypedDict

# 定义状态结构
class RestaurantState(TypedDict):
    order: str          # 客人的订单
    is_valid: bool      # 订单是否有效
    dish: str           # 做好的菜

# 函数节点1：接收订单
def take_order(state: RestaurantState) -> RestaurantState:
    """服务员接收订单"""
    print(f"📋 收到订单：{state['order']}")
    return state

# 函数节点2：验证订单
def validate_order(state: RestaurantState) -> RestaurantState:
    """验证订单是否有效"""
    # 假设菜单上有这些菜
    menu = ["宫保鸡丁", "鱼香肉丝", "麻婆豆腐"]
    state["is_valid"] = state["order"] in menu
    if state["is_valid"]:
        print(f"✅ 订单有效：{state['order']}")
    else:
        print(f"❌ 无效订单：{state['order']} 不在菜单上")
    return state

# 构建图
builder = StateGraph(RestaurantState)
builder.add_node("take_order", take_order)
builder.add_node("validate_order", validate_order)

# 设置边（连接节点）
builder.set_entry_point("take_order")
builder.add_edge("take_order", "validate_order")
builder.add_edge("validate_order", "__end__")

# 编译并运行
graph = builder.compile()
result = graph.invoke({"order": "宫保鸡丁"})

这个例子展示了最基础的函数节点用法。我们定义了两个节点：接收订单和验证订单，它们通过状态传递信息。

2.2 LLM节点（LLM Node）——AI的大脑

LLM节点是LangGraph的核心亮点之一。它将大语言模型（如GPT-4、Claude、文心一言等）封装成一个Node，让AI能够在工作流中”思考”和”决策”。

适用场景：

• 理解自然语言输入
• 生成文本回复
• 进行推理和决策
• 提取结构化信息
• 多轮对话管理

代码示例：

from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
from typing import TypedDict
import os

# 设置OpenAI API密钥（请替换为你的密钥）
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 定义状态
class ChatState(TypedDict):
    message: str
    response: str

# 创建LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# LLM节点
def ai_respond(state: ChatState) -> ChatState:
    """AI回复节点"""
    # 调用LLM生成回复
    response = llm.invoke(state["message"])
    state["response"] = response.content
    print(f"🤖 AI：{state['response']}")
    return state

# 构建图
builder = StateGraph(ChatState)
builder.add_node("ai_respond", ai_respond)
builder.set_entry_point("ai_respond")
builder.add_edge("ai_respond", "__end__")

graph = builder.compile()
result = graph.invoke({"message": "你好，请介绍一下LangGraph"})

在这个例子中，ai_respond节点封装了对GPT模型的调用。当状态传递到这个节点时，它会将用户的message发送给LLM，并将回复存入response字段。

2.3 工具节点（Tool Node）——连接外部世界的桥梁

工具节点是LangGraph的”超能力”来源。它允许AI调用外部工具，比如查询天气、搜索网络、操作数据库、发送邮件等。这让AI从”只会聊天”变成了”能够行动”。

适用场景：

• 获取实时信息（天气、股价、新闻）
• 执行计算或数据分析
• 操作外部系统（数据库、API）
• 搜索互联网
• 调用专业软件或服务

代码示例：

from langgraph.graph import StateGraph
from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from typing import TypedDict

# 定义状态
class ToolState(TypedDict):
    query: str
    result: str

# 定义工具（使用装饰器语法）
@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气"""
    # 这里应该是真实的天气API调用
    # 为演示方便，返回模拟数据
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，28°C",
        "深圳": "小雨，30°C"
    }
    return weather_data.get(city, f"抱歉，暂无{city}的天气信息")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)  # 注意：生产环境要小心安全风险
        return f"计算结果：{result}"
    except:
        return "计算失败，请检查表达式"

# 工具列表
tools = [search_weather, calculate]

# 创建工具节点
tool_node = ToolNode(tools)

# 简单的调用示例
def use_tool(state: ToolState) -> ToolState:
    """使用工具的节点"""
    # 根据query决定调用哪个工具
    if "天气" in state["query"]:
        city = state["query"].replace("天气", "").strip()
        state["result"] = search_weather.invoke(city)
    elif "计算" in state["query"]:
        expr = state["query"].replace("计算", "").strip()
        state["result"] = calculate.invoke(expr)
    else:
        state["result"] = "我不确定你想做什么，可以试试查询天气或计算"
    return state

# 构建图
builder = StateGraph(ToolState)
builder.add_node("use_tool", use_tool)
builder.set_entry_point("use_tool")
builder.add_edge("use_tool", "__end__")

graph = builder.compile()

# 测试
result = graph.invoke({"query": "北京天气"})
print(result["result"])

工具节点的强大之处在于，它让AI不再局限于自己的”知识库”，而是能够动态地获取信息、执行操作。这是构建真正有用AI应用的关键。

三、Node间的数据传递——State的秘密

现在你已经了解了三种Node类型，但可能还有一个疑问：Node之间是如何传递信息的？

答案是：State（状态）。

3.1 什么是State？

State是一个共享的数据容器，所有的Node都可以读取和修改它。你可以把它想象成餐厅里的订单小票——服务员写下客人的需求，厨师按照小票做菜，收银员根据小票结账，每个人都看得到同一张小票上的信息。

在LangGraph中，State通常用TypedDict来定义，这样既能明确字段类型，又能享受类型提示的好处。

from typing import TypedDict, List

class ConversationState(TypedDict):
    user_name: str           # 用户名
    messages: List[str]      # 对话历史
    current_intent: str      # 当前意图
    extracted_entities: dict # 提取的实体

3.2 从State读取数据

每个Node的函数签名都接收state作为参数，你可以直接访问其中的字段：

def read_example(state: ConversationState) -> ConversationState:
    # 读取用户名
    name = state["user_name"]
    
    # 读取消息列表
    messages = state["messages"]
    
    # 检查字段是否存在（使用get方法）
    intent = state.get("current_intent", "未知")
    
    print(f"用户{name}说了：{messages[-1] if messages else '无消息'}")
    return state

3.3 向State写入数据

Node可以修改State，这些修改会自动传递给下一个Node：

def write_example(state: ConversationState) -> ConversationState:
    # 添加新消息
    state["messages"].append("这是新消息")
    
    # 更新意图
    state["current_intent"] = "查询天气"
    
    # 添加新实体
    state["extracted_entities"]["city"] = "北京"
    
    return state

3.4 重要注意事项

State是 mutable（可变的）：虽然你可以直接修改state，但建议始终返回修改后的state，这是一个好的编程习惯。

字段默认值：如果某个字段可能不存在，使用get方法提供默认值，避免KeyError。

# 好的做法
city = state.get("city", "北京")  # 如果没有city，默认用北京

# 有风险的做法
city = state["city"]  # 如果city不存在，会抛出KeyError

四、实战：天气查询助手

理论讲得再多，不如动手实践。下面我们来构建一个完整的天气查询助手，它结合了LLM节点和工具节点，能够：

1. 理解用户的自然语言查询
2. 提取城市名称
3. 调用天气API获取实时天气
4. 生成友好的回复

4.1 完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
LangGraph 天气查询助手
结合LLM节点和工具节点，实现智能天气查询
"""

import os
from typing import TypedDict, Annotated, Sequence
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import json

# ========== 1. 配置部分 ==========
# 请确保设置了OPENAI_API_KEY环境变量
# export OPENAI_API_KEY="your-api-key"

# ========== 2. 定义工具 ==========
@tool
def get_weather(city: str) -> str:
    """
    查询指定城市的天气信息。
    
    Args:
        city: 城市名称，如"北京"、"上海"、"纽约"
    
    Returns:
        该城市的天气信息，包括温度、天气状况等
    """
    # 模拟天气API（实际项目中替换为真实的天气API）
    weather_db = {
        "北京": {"temp": 22, "condition": "晴朗", "humidity": 45, "wind": "北风3级"},
        "上海": {"temp": 26, "condition": "多云", "humidity": 65, "wind": "东南风2级"},
        "广州": {"temp": 30, "condition": "小雨", "humidity": 80, "wind": "南风3级"},
        "深圳": {"temp": 31, "condition": "雷阵雨", "humidity": 85, "wind": "东南风4级"},
        "杭州": {"temp": 24, "condition": "阴天", "humidity": 55, "wind": "东风2级"},
        "成都": {"temp": 20, "condition": "多云", "humidity": 70, "wind": "无持续风向"},
        "武汉": {"temp": 25, "condition": "晴天", "humidity": 50, "wind": "北风2级"},
        "西安": {"temp": 23, "condition": "晴朗", "humidity": 40, "wind": "西北风3级"},
        "纽约": {"temp": 15, "condition": "Rainy", "humidity": 70, "wind": "NW 5mph"},
        "伦敦": {"temp": 12, "condition": "Cloudy", "humidity": 75, "wind": "SW 8mph"},
        "东京": {"temp": 18, "condition": "Sunny", "humidity": 60, "wind": "E 4mph"},
    }
    
    city = city.strip()
    if city in weather_db:
        data = weather_db[city]
        return f"{city}当前天气：{data['condition']}，温度{data['temp']}°C，湿度{data['humidity']}%，{data['wind']}"
    else:
        return f"抱歉，暂时无法获取{city}的天气信息。目前支持的城市：{', '.join(list(weather_db.keys())[:8])}等"

@tool
def get_weather_advice(condition: str, temp: int) -> str:
    """
    根据天气状况给出穿衣和出行建议。
    
    Args:
        condition: 天气状况，如"晴朗"、"下雨"、"多云"
        temp: 温度（摄氏度）
    
    Returns:
        穿衣和出行建议
    """
    advice = []
    
    # 穿衣建议
    if temp >= 30:
        advice.append("天气炎热，建议穿轻薄透气的衣物，注意防晒")
    elif temp >= 20:
        advice.append("天气舒适，穿短袖或薄长袖都可以")
    elif temp >= 10:
        advice.append("天气较凉，建议穿外套或薄毛衣")
    else:
        advice.append("天气寒冷，请穿厚外套，注意保暖")
    
    # 出行建议
    if "雨" in condition or "Rain" in condition:
        advice.append("有雨，出门记得带伞，路面湿滑请注意安全")
    elif "晴" in condition or "Sunny" in condition:
        advice.append("天气晴好，适合户外活动")
    elif "雪" in condition:
        advice.append("有雪，路面可能结冰，驾车请注意安全")
    
    return "；".join(advice)

# 工具列表
tools = [get_weather, get_weather_advice]

# ========== 3. 定义状态 ==========
class WeatherState(TypedDict):
    """天气查询助手的状态"""
    messages: Annotated[Sequence[BaseMessage], "对话消息列表"]
    city: Annotated[str, "提取的城市名"]
    weather_info: Annotated[str, "天气信息"]
    final_response: Annotated[str, "最终回复"]

# ========== 4. 创建LLM并绑定工具 ==========
# 初始化语言模型
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
)

# 将工具绑定到LLM
llm_with_tools = llm.bind_tools(tools)

# ========== 5. 定义节点函数 ==========

def extract_city_node(state: WeatherState) -> WeatherState:
    """
    节点1：从用户输入中提取城市名
    使用LLM理解用户意图并提取关键信息
    """
    # 获取最新的用户消息
    last_message = state["messages"][-1].content if state["messages"] else ""
    
    # 使用简单的关键词提取（实际项目可以用更复杂的NLP）
    # 这里为了演示，我们用一个简单的提示让LLM提取
    extraction_prompt = f"""
    从以下句子中提取城市名称，只返回城市名，不要其他内容：
    "{last_message}"
    
    如果找不到城市名，返回"未知"。
    """
    
    response = llm.invoke(extraction_prompt)
    city = response.content.strip().replace(" ", "").replace("市", "")
    
    # 过滤掉常见的非城市词
    if city in ["未知", "Unknown", "没有", "无"]:
        city = ""
    
    state["city"] = city
    print(f"🎯 提取到城市：{city if city else '未识别'}")
    
    return state

def agent_node(state: WeatherState) -> WeatherState:
    """
    节点2：AI智能体节点
    决定是调用工具还是直接回复用户
    """
    messages = state["messages"]
    
    # 添加系统提示
    system_message = HumanMessage(content="""你是一个友好的天气助手。当用户询问天气时：
1. 首先确认城市名
2. 调用get_weather工具获取天气
3. 根据天气状况，调用get_weather_advice获取建议
4. 用温暖友好的语气回复用户

如果用户没有指定城市，请礼貌地询问。""")
    
    # 合并消息
    all_messages = [system_message] + list(messages)
    
    # 调用LLM
    response = llm_with_tools.invoke(all_messages)
    
    # 将AI回复添加到消息列表
    state["messages"] = list(messages) + [response]
    
    print(f"🤖 AI决策：{'调用工具' if response.tool_calls else '直接回复'}")
    
    return state

def tool_executor_node(state: WeatherState) -> WeatherState:
    """
    节点3：工具执行节点
    执行AI决定调用的工具
    """
    messages = state["messages"]
    last_message = messages[-1]
    
    # 检查最后一条消息是否是工具调用请求
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        tool_messages = []
        
        for tool_call in last_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            
            print(f"🔧 执行工具：{tool_name}，参数：{tool_args}")
            
            # 查找并执行对应的工具
            for tool in tools:
                if tool.name == tool_name:
                    try:
                        result = tool.invoke(tool_args)
                        tool_messages.append(ToolMessage(
                            content=result,
                            tool_call_id=tool_call["id"]
                        ))
                        
                        # 如果是天气查询，保存到state
                        if tool_name == "get_weather":
                            state["weather_info"] = result
                            
                    except Exception as e:
                        tool_messages.append(ToolMessage(
                            content=f"工具执行出错：{str(e)}",
                            tool_call_id=tool_call["id"]
                        ))
                    break
        
        # 添加工具执行结果到消息列表
        state["messages"] = list(messages) + tool_messages
    
    return state

def generate_response_node(state: WeatherState) -> WeatherState:
    """
    节点4：生成最终回复
    整合所有信息，生成友好的回复
    """
    messages = state["messages"]
    
    # 调用LLM生成最终回复
    final_prompt = """基于以上对话，请生成一个温暖、友好的天气回复。
包含天气信息和生活建议，语气要像朋友一样亲切。"""
    
    final_messages = list(messages) + [HumanMessage(content=final_prompt)]
    response = llm.invoke(final_messages)
    
    state["final_response"] = response.content
    print(f"✅ 生成回复：{response.content[:50]}...")
    
    return state

def check_city_node(state: WeatherState) -> str:
    """
    条件判断节点：检查是否识别到城市
    返回下一个节点的名称
    """
    if state.get("city"):
        return "agent"
    else:
        return "ask_for_city"

def ask_for_city_node(state: WeatherState) -> WeatherState:
    """
    询问用户城市名的节点
    """
    state["final_response"] = "请问您想查询哪个城市的天气呢？比如北京、上海、广州等。"
    print("❓ 未识别城市，询问用户")
    return state

def should_continue(state: WeatherState) -> str:
    """
    条件边：判断是否需要继续调用工具
    """
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果最后一条消息是AI的回复且包含工具调用，继续执行工具
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "tools"
    
    # 否则结束
    return "end"

# ========== 6. 构建图 ==========
# 创建图构建器
builder = StateGraph(WeatherState)

# 添加节点
builder.add_node("extract_city", extract_city_node)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_executor_node)
builder.add_node("generate_response", generate_response_node)
builder.add_node("ask_for_city", ask_for_city_node)

# 设置入口点
builder.set_entry_point("extract_city")

# 添加边
builder.add_conditional_edges(
    "extract_city",
    check_city_node,
    {
        "agent": "agent",
        "ask_for_city": "ask_for_city"
    }
)

builder.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": "generate_response"
    }
)

builder.add_edge("tools", "agent")  # 工具执行后返回给agent
builder.add_edge("ask_for_city", END)
builder.add_edge("generate_response", END)

# 编译图
graph = builder.compile()

# ========== 7. 运行示例 ==========
def chat_with_weather_assistant(user_input: str):
    """
    与天气助手对话的便捷函数
    """
    print("\n" + "="*50)
    print(f"👤 用户：{user_input}")
    print("="*50)
    
    # 初始化状态
    initial_state = {
        "messages": [HumanMessage(content=user_input)],
        "city": "",
        "weather_info": "",
        "final_response": ""
    }
    
    # 运行图
    result = graph.invoke(initial_state)
    
    print("\n📤 最终回复：")
    print(result["final_response"])
    print("="*50)
    
    return result["final_response"]

# ========== 8. 主程序 ==========
if __name__ == "__main__":
    # 测试对话
    print("🌤️  LangGraph 天气查询助手 🌤️")
    print("输入 'quit' 退出程序\n")
    
    # 预设测试
    test_inputs = [
        "今天北京天气怎么样？",
        "上海明天穿什么合适？",
        "帮我查一下深圳的天气"
    ]
    
    print("📝 运行预设测试...\n")
    for test_input in test_inputs:
        chat_with_weather_assistant(test_input)
    
    print("\n💡 提示：可以修改代码中的 test_inputs 列表来测试其他输入")

4.2 代码逐行讲解

第1-10行：导入和配置

• 导入必要的库，包括LangGraph的核心组件和OpenAI的客户端
• 设置环境变量用于存储API密钥（安全考虑，不要硬编码）

第12-63行：定义工具

• 使用@tool装饰器定义两个工具：get_weather和get_weather_advice
• 每个工具都有清晰的文档字符串，说明参数和返回值
• 工具的本质就是函数，可以是任何Python逻辑

第65-72行：定义State

• 使用TypedDict定义状态结构
• Annotated用于添加类型元数据，让代码更易读
• State包含：消息列表、城市名、天气信息、最终回复

第74-82行：初始化LLM

• 创建ChatOpenAI实例，指定模型和温度参数
• 使用bind_tools将工具”绑定”到LLM，让LLM知道可以调用哪些工具

第84-203行：定义节点函数
这是整个程序的核心，定义了6个节点：

• extract_city_node：从用户输入中提取城市名
• agent_node：AI决策节点，决定调用工具还是直接回复
• tool_executor_node：执行AI调用的工具
• generate_response_node：生成最终友好回复
• ask_for_city_node：当没有识别到城市时询问用户
• check_city_node和should_continue：条件判断函数，决定流程走向

第205-245行：构建图

• 创建StateGraph实例
• 使用add_node添加所有节点
• 使用set_entry_point设置入口
• 使用add_edge和add_conditional_edges连接节点
• 最后用compile编译成可执行的图

第247-290行：运行和测试

• chat_with_weather_assistant函数封装了完整的对话流程
• 主程序中运行预设测试用例

4.3 运行结果示例

🌤️  LangGraph 天气查询助手 🌤️
输入 'quit' 退出程序

📝 运行预设测试...

==================================================
👤 用户：今天北京天气怎么样？
==================================================
🎯 提取到城市：北京
🤖 AI决策：调用工具
🔧 执行工具：get_weather，参数：{'city': '北京'}
🤖 AI决策：调用工具
🔧 执行工具：get_weather_advice，参数：{'condition': '晴朗', 'temp': 22}
✅ 生成回复：今天北京天气晴朗，温度22°C，...

📤 最终回复：
你好！北京今天天气真不错呢～ ☀️
...
==================================================

五、Node设计最佳实践

通过上面的实战，相信你对Node有了更深的理解。下面分享一些设计Node的最佳实践，帮助你写出更好的代码。

5.1 单一职责原则

每个Node只做一件事。不要把太多逻辑塞到一个Node里。

❌ 不好的做法：

def messy_node(state):
    # 解析输入
    # 调用API
    # 处理数据
    # 生成回复
    # 保存日志
    pass

✅ 好的做法：

def parse_input(state): pass
def call_api(state): pass
def process_data(state): pass
def generate_reply(state): pass
def save_log(state): pass

5.2 完善的错误处理

Node可能遇到各种错误，要做好防护：

def robust_node(state: MyState) -> MyState:
    try:
        result = some_risky_operation()
        state["result"] = result
    except NetworkError:
        state["error"] = "网络连接失败，请检查网络"
    except TimeoutError:
        state["error"] = "请求超时，请稍后重试"
    except Exception as e:
        state["error"] = f"未知错误：{str(e)}"
        # 记录详细错误日志
        logger.error(f"robust_node failed: {e}", exc_info=True)
    
    return state

5.3 添加日志记录

日志是调试的利器，特别是在复杂的图流程中：

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def logged_node(state):
    logger.info(f"Node started with state: {state}")
    
    # 执行逻辑...
    
    logger.info(f"Node finished, output: {result}")
    return state

5.4 使用类型提示

类型提示不仅能让代码更易读，还能在IDE中享受自动补全：

from typing import TypedDict, Literal

class State(TypedDict):
    query: str
    status: Literal["pending", "success", "error"]
    result: dict | None

def typed_node(state: State) -> State:
    # IDE会提示 state 有哪些字段
    pass

5.5 保持State的扁平化

尽量避免过深的嵌套结构，让State保持扁平：

❌ 避免：

{"user": {"profile": {"name": "", "age": 0}}}

✅ 推荐：

{"user_name": "", "user_age": 0}

5.6 命名要清晰

Node和State字段的命名要见名知意：

❌ 避免：

builder.add_node("n1", f1)
state["d"]

✅ 推荐：

builder.add_node("extract_keywords", extract_keywords)
state["extracted_keywords"]

六、总结与下篇预告

6.1 本文回顾

通过这篇文章，我们学习了：

1. 什么是Node：LangGraph的基本构建单元，就像餐厅里的不同角色
2. 三种Node类型：
   • 函数节点：灵活的基础砖块
   • LLM节点：AI的大脑
   • 工具节点：连接外部世界的桥梁
3. 数据传递：通过State在Node间共享数据
4. 实战项目：完整的天气查询助手，展示了如何组合不同类型的Node
5. 最佳实践：单一职责、错误处理、日志记录等

6.2 核心要点

• Node是LangGraph的核心概念，掌握Node就掌握了构建复杂AI应用的基础
• 选择合适的Node类型：简单逻辑用函数节点，需要AI决策用LLM节点，需要外部能力用工具节点
• State是Node间的”通信协议”，设计好State结构很重要
• 实际项目中，Node的设计要遵循单一职责原则，做好错误处理

6.3 下篇预告

在下一篇文章中，我们将深入探讨Tool Calling——

如果说Node是餐厅里的员工，那么Edge就是员工之间传递信息的通道和规则。我们会学习：

• 什么是条件边（Conditional Edge）
• 如何实现循环和分支逻辑
• 如何构建多轮对话
• 实战：构建一个更复杂的多Agent协作系统

敬请期待Tool Calling：让AI拥有”超能力”！

---

参考资源：

• LangGraph官方文档
• LangChain官方教程
• 项目GitHub仓库

希望这篇文章能帮助你入门LangGraph！如果有任何问题，欢迎在评论区留言交流。

---

📚 LangGraph 入门系列

本系列共10篇文章，带你从零掌握 LangGraph：

篇目	标题	链接
01	LangGraph 是什么？	阅读
02	State 状态管理	阅读
03	Node 节点	阅读
04	Tool Calling	阅读
05	Memory 记忆系统	阅读
06	HITL 人机协作	阅读
07	Multi-Agent	阅读
08	Streaming 流式输出	阅读
09	Conditional Edges	阅读
10	Production 生产部署	阅读

当前位置： 第3篇 / 共10篇

• 上一篇：State 状态管理
• 下一篇：Tool Calling