Node 与 Edge
Node（节点）和 Edge（边）是 StateGraph 的两个基本构建单元。节点是执行的单元，边定义了执行的顺序和条件。
Node（节点）
基本节点
节点就是一个函数，接收当前的 state，返回 state 的更新：
python
def my_node(state: State) -> dict:
# 读取 state
input_value = state["input"]
# 执行逻辑
result = f"处理: {input_value}"
# 返回更新（只返回需要修改的字段）
return {"output": result}节点签名
节点函数可以接受以下参数：
python
def my_node(
state: State, # 必需的 state 参数
config: RunnableConfig, # 运行配置（可选）
**kwargs, # 其他参数
) -> dict:
pass带 Config 的节点
python
from langchain_core.runnables import RunnableConfig
def node_with_config(state: State, config: RunnableConfig):
thread_id = config["configurable"]["thread_id"]
print(f"Running on thread: {thread_id}")
return {"output": f"done on {thread_id}"}异步节点
python
async def async_node(state: State):
result = await some_async_function(state["input"])
return {"output": result}多返回值
节点可以一次更新多个 state 字段：
python
def multi_output_node(state: State):
return {
"intermediate": "处理中...",
"status": "processing",
"timestamp": time.time(),
}Edge（边）
普通边：无条件流转
python
from langgraph.graph import StateGraph, START, END
builder = StateGraph(State)
# 添加节点
builder.add_node("step1", step1_function)
builder.add_node("step2", step2_function)
# 定义执行顺序
builder.add_edge(START, "step1") # 入口 → step1
builder.add_edge("step1", "step2") # step1 → step2
builder.add_edge("step2", END) # step2 → 结束条件边：根据条件路由
python
from typing import Literal
def router(state: State) -> Literal["tool_node", "respond", END]:
"""根据最后一条消息的内容决定下一步"""
last_message = state["messages"][-1]
if last_message.get("tool_calls"):
return "tool_node" # 有工具调用 → 执行工具
elif state.get("need_clarification"):
return "respond" # 需要澄清 → 回复用户
else:
return END # 完成 → 结束
builder.add_conditional_edges(
"llm_node", # 起始节点
router, # 路由函数
{ # 路由图
"tool_node": "tool_node",
"respond": "respond",
END: END,
}
)也可以简写为列表形式：
python
builder.add_conditional_edges(
"llm_node",
router,
["tool_node", "respond", END]
)条件边中 END 的引用方式：
- 传入
END常量（已导入）：["tool_node", END] - 传入字符串
"__end__"：["tool_node", "__end__"] - 传入字典：
{... END: END}或{... "__end__": END}
并行边
多个节点可以并行执行，LangGraph 会自动处理：
python
# START 后同时执行 A 和 B
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
# A 和 B 都完成后执行 C
builder.add_edge("node_a", "node_c")
builder.add_edge("node_b", "node_c")图结构示例
顺序执行
START → A → B → C → ENDpython
builder.add_edge(START, "A")
builder.add_edge("A", "B")
builder.add_edge("B", "C")
builder.add_edge("C", END)条件分支
→ tool_node ┐
START → LLM → END
→ respond ─┘python
builder.add_node("LLM", llm_node)
builder.add_node("tool_node", tool_node)
builder.add_node("respond", respond_node)
builder.add_edge(START, "LLM")
builder.add_conditional_edges("LLM", router, ["tool_node", "respond", END])
builder.add_edge("tool_node", "LLM") # 工具执行完后回到 LLM
builder.add_edge("respond", END)并行执行
→ A ┐
START → C → END
→ B ┘python
builder.add_edge(START, "A")
builder.add_edge(START, "B")
builder.add_edge("A", "C")
builder.add_edge("B", "C")
builder.add_edge("C", END)节点命名规范
- 使用小写字母 + 下划线（snake_case）
- 名称应描述节点的功能：
retrieve_docs、call_llm、parse_output - 避免特殊字符
- 在同一张图中保持名称风格一致
最佳实践
- 节点粒度适中：太细（每行代码一个节点）可读性差，太粗（一个大函数一个节点）丧失编排优势
- 无副作用：节点函数不应该有副作用（如修改全局变量），所有状态变化通过返回值体现
- 幂等性：尽量让节点函数在重复执行时结果一致
- 命名清晰：好的节点名胜过注释