LangGraph 核心之节点（Node）

LangGraph 的核心在于其简洁而强大的图计算模型，这一模型的基石由四个核心原语构成：状态（State）、节点（Node）、边（Edge），以及命令（Command）。
理解这四个原语的概念及其相互作用方式，是掌握 LangGraph 并构建复杂智能体系统的重中之重。可以将这四个原语比作乐高积木最基本的、也是最核心的模块，理解了它们，就如同掌握了乐高搭建的"语言"，后续才能使用更高级的技巧，搭建出各种各样精巧、复杂、功能强大的智能体系统。

本章将深入探讨 LangGraph 核心原语之节点（Node）

节点（Node）

节点是 LangGraph 图结构中的基本计算单元。每一个节点都封装了一个独立的计算逻辑，例如调用语言模型、执行工具、进行条件判断、或者仅仅是一个简单的数据处理函数。

在 LangGraph 中，节点本质上就是一个 Python 函数。这个函数接收当前的状态作为输入，并返回一个新的状态（或者状态的更新部分）作为输出。

示例 1-1：节点函数的基本结构

def my_node(state):
    """
    节点函数示例
    """
    # 从状态中读取数据
    input_data = state.get("some_key", "default_value")
    
    # 执行节点计算逻辑
    def process_data(data):
        return f"处理后的数据: {data.upper()}"
    
    output_data = process_data(input_data)
    
    # 返回新的状态（或状态的更新部分）
    return {"some_key": output_data, "another_key": "new_value"}

# 测试节点函数
print("测试节点函数：")
test_state = {"some_key": "hello world", "existing_key": "existing_value"}
print(f"输入状态: {test_state}")

result = my_node(test_state)
print(f"节点输出: {result}")

# 模拟状态更新（LangGraph会自动处理状态合并）
updated_state = {**test_state, **result}
print(f"更新后状态: {updated_state}")

测试节点函数：
输入状态: {'some_key': 'hello world', 'existing_key': 'existing_value'}
节点输出: {'some_key': '处理后的数据: HELLO WORLD', 'another_key': 'new_value'}
更新后状态: {'some_key': '处理后的数据: HELLO WORLD', 'existing_key': 'existing_value', 'another_key': 'new_value'}

示例 1-2：一个包含 LLM 节点的 LangGraph 图

from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 定义状态结构体 
class ChatState(MessagesState):
    user_question: str  # 用户问题
    llm_response: str   # LLM回复

# 定义 LLM 节点 
def llm_node(state):
    prompt = ChatPromptTemplate.from_messages([
        ("human", "{question}")
    ])
    model = ChatOpenAI(model="Qwen/Qwen3-8B")
    chain = prompt | model
    
    response = chain.invoke({"question": state['user_question']}).content
    return {"llm_response": response}

# 构建图 
builder = StateGraph(ChatState)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)
graph = builder.compile()

print("LangGraph 图构建完成")
print("节点: llm_node")
print("边: START -> llm_node -> END")

# 测试图的执行
print("\n测试图执行：")
try:
    result = graph.invoke({"user_question": "你好，LangGraph！"})
    print(f"执行结果: {result}")
except Exception as e:
    print(f"需要配置API密钥才能实际运行LLM: {e}")
    print("图结构已成功创建，可以在配置API后运行")

LangGraph 图构建完成
节点: llm_node
边: START -> llm_node -> END

测试图执行：
执行结果: {'messages': [], 'user_question': '你好，LangGraph！', 'llm_response': '\n\n你好！我是通义千问，很高兴认识你。LangGraph 是 LangChain 的一个扩展库，它提供了一种通过图结构来组织和操作数据的方式，使得构建更复杂的 AI 应用（如知识图谱、问答系统、推荐系统等）变得更加简单和高效。如果你有任何关于 LangGraph 的问题，或者需要帮助构建某个具体的应用，欢迎随时告诉我！'}

示例 1-3：为 LangGraph 节点配置重试策略的代码示例

import operator
import sqlite3
import random
import time
from typing import Annotated, Sequence
from typing_extensions import TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

# 模拟数据库类
class MockSQLDatabase:
    def __init__(self):
        self.connection_stable = False
        self.call_count = 0

    def run(self, query):
        self.call_count += 1
        print(f"🗄️ 数据库查询 (第{self.call_count}次): {query}")

        # 模拟不稳定的数据库连接 - 前2次调用会失败
        if self.call_count <= 2:
            print(f"❌ 数据库连接失败 (模拟错误)")
            raise sqlite3.OperationalError("数据库连接超时")

        print(f"✅ 数据库查询成功")
        return "艺术家数据: Van Gogh, Picasso, Da Vinci, Monet, Renoir"

# 模拟 LLM 类
class MockChatOpenAI:
    def __init__(self, model="mock-model"):
        self.model = model
        self.call_count = 0

    def invoke(self, messages):
        self.call_count += 1
        print(f"🤖 LLM调用 (第{self.call_count}次)")

        # 模拟 LLM 偶尔失败 - 30% 概率失败
        if random.random() < 0.3:
            print(f"❌ LLM服务暂时不可用 (模拟错误)")
            raise ConnectionError("LLM服务连接失败")

        last_message = messages[-1] if messages else None
        content = f"基于查询结果，我为您找到了相关的艺术家信息。这是第{self.call_count}次成功调用的响应。"
        print(f"✅ LLM响应生成成功")
        return AIMessage(content=content)

# 初始化模拟组件
db = MockSQLDatabase()
model = MockChatOpenAI(model="Mock-GPT-4")

# 定义图的状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

def query_database(state):
    """查询数据库节点 - 配置了特定异常重试"""
    print(f"\n📊 执行数据库查询节点...")
    query_result = db.run("SELECT * FROM Artist LIMIT 10;")
    return {"messages": [AIMessage(content=f"数据库查询结果: {query_result}")]}

def call_model(state):
    """调用模型节点 - 配置了最大重试次数"""
    print(f"\n🧠 执行模型调用节点...")
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def user_input_node(state):
    """用户输入节点"""
    print(f"\n👤 添加用户输入...")
    user_message = HumanMessage(content="请帮我查询一些著名艺术家的信息")
    print(f"📝 用户问题: {user_message.content}")
    return {"messages": [user_message]}

# 定义图 builder
print("🏗️ 构建带重试策略的 LangGraph...")
builder = StateGraph(AgentState)

# 添加用户输入节点
builder.add_node("user_input", user_input_node)

# 为 call_model 节点配置重试策略: 最大重试 5 次，包含退避策略
builder.add_node(
    "model",
    call_model,
    retry=RetryPolicy(
        max_attempts=5,           # 最大重试5次
        initial_interval=0.5,     # 初始重试间隔0.5秒
        backoff_factor=2.0,       # 退避因子2.0 (指数退避)
        max_interval=8.0,         # 最大重试间隔8秒
        jitter=True              # 添加随机抖动
    )
)

# 为 query_database 节点配置重试策略: 针对 sqlite3.OperationalError 异常进行重试
builder.add_node(
    "query_database",
    query_database,
    retry=RetryPolicy(
        retry_on=sqlite3.OperationalError,  # 只对数据库操作错误重试
        max_attempts=4,                     # 最大重试4次
        initial_interval=1.0,               # 初始间隔1秒
        backoff_factor=1.5                  # 较小的退避因子
    )
)

# 定义边
builder.add_edge(START, "user_input")
builder.add_edge("user_input", "model")
builder.add_edge("model", "query_database")
builder.add_edge("query_database", END)

# 编译图
graph = builder.compile()
print("✅ 图构建完成！")

# 测试运行
print("\n=== 🚀 重试策略演示 ===")
print("📋 测试场景:")
print("  - 数据库节点: 前2次调用会失败，第3次成功")
print("  - 模型节点: 30% 概率失败，会自动重试")
print("  - 两个节点都配置了不同的重试策略\n")

try:
    # 运行图
    result = graph.invoke({"messages": []})

    print(f"\n=== ✨ 执行完成 ===")
    print(f"📊 最终消息数量: {len(result['messages'])}")
    for i, msg in enumerate(result['messages']):
        print(f"  {i+1}. [{msg.__class__.__name__}] {msg.content[:60]}...")

    print(f"\n=== 📈 重试统计 ===")
    print(f"🗄️ 数据库调用次数: {db.call_count}")
    print(f"🤖 模型调用次数: {model.call_count}")

except Exception as e:
    print(f"\n❌ 执行失败: {e}")
    print(f"🗄️ 数据库调用次数: {db.call_count}")
    print(f"🤖 模型调用次数: {model.call_count}")


🏗️ 构建带重试策略的 LangGraph...
✅ 图构建完成！

=== 🚀 重试策略演示 ===
📋 测试场景:
  - 数据库节点: 前2次调用会失败，第3次成功
  - 模型节点: 30% 概率失败，会自动重试
  - 两个节点都配置了不同的重试策略


👤 添加用户输入...
📝 用户问题: 请帮我查询一些著名艺术家的信息

🧠 执行模型调用节点...
🤖 LLM调用 (第1次)
✅ LLM响应生成成功

📊 执行数据库查询节点...
🗄️ 数据库查询 (第1次): SELECT * FROM Artist LIMIT 10;
❌ 数据库连接失败 (模拟错误)

📊 执行数据库查询节点...
🗄️ 数据库查询 (第2次): SELECT * FROM Artist LIMIT 10;
❌ 数据库连接失败 (模拟错误)

📊 执行数据库查询节点...
🗄️ 数据库查询 (第3次): SELECT * FROM Artist LIMIT 10;
✅ 数据库查询成功

=== ✨ 执行完成 ===
📊 最终消息数量: 3
  1. [HumanMessage] 请帮我查询一些著名艺术家的信息...
  2. [AIMessage] 基于查询结果，我为您找到了相关的艺术家信息。这是第1次成功调用的响应。...
  3. [AIMessage] 数据库查询结果: 艺术家数据: Van Gogh, Picasso, Da Vinci, Monet, Renoir...

=== 📈 重试统计 ===
🗄️ 数据库调用次数: 3
🤖 模型调用次数: 1

1. RetryPolicy 配置：
   • model 节点：通用重试策略，处理各种异常
   • query_database 节点：针对特定数据库异常的重试策略
2. 不同的重试参数：
   • 最大重试次数、初始间隔、退避因子等
   • 展示指数退避和抖动机制
3. 模拟失败场景：
   • 数据库连接不稳定（前几次必然失败）
   • LLM 服务偶尔不可用（随机失败）
4. 重试效果演示：
   • 显示每次重试的过程
   • 统计实际调用次数
   • 展示重试策略的实际效果

总结

在LangGraph的拓扑结构中，节点是承载具体业务逻辑的最小单元，相当于工作流中的“步骤”或“函数”。它的核心职责是：接收输入数据、执行预设逻辑（如调用LLM、处理数据、调用外部API等），并输出处理结果，为后续节点或流程分支提供数据支撑。

参考文献和资料
• LangGraph文档：https://docs.langchain.com/oss/python/langgraph/overview
• 《LangGraph实战》

微信公众号（关注获取最新资讯）：