节点与边 #
概述 #
节点（Nodes）和边（Edges）是 LangGraph 图结构的两个基本构建块。节点执行工作，边决定下一步做什么。通过组合节点和边，你可以创建复杂的、有状态的、可循环的工作流。
text
┌─────────────────────────────────────────────────────────────┐
│ 节点与边的关系 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 节点（Nodes）： │
│ - 执行实际工作 │
│ - 接收状态，返回更新 │
│ - 可以是函数或 Runnable │
│ │
│ 边（Edges）： │
│ - 定义控制流 │
│ - 决定下一个执行的节点 │
│ - 可以是固定或条件的 │
│ │
│ 简而言之： │
│ 节点做工作，边告诉下一步做什么 │
│ │
└─────────────────────────────────────────────────────────────┘
节点（Nodes） #
节点函数签名 #
节点是 Python 函数，可以接收以下参数：
python
from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
from typing_extensions import TypedDict
class State(TypedDict):
messages: list
count: int
def my_node(
state: State, # 必需：当前状态
config: RunnableConfig, # 可选：配置信息
runtime: Runtime, # 可选：运行时上下文
):
return {"count": state["count"] + 1}
基本节点 #
python
def agent_node(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response]}
graph.add_node("agent", agent_node)
带配置的节点 #
python
def node_with_config(state: State, config: RunnableConfig):
thread_id = config["configurable"]["thread_id"]
user_id = config["configurable"].get("user_id")
return {"thread_id": thread_id, "user_id": user_id}
graph.add_node("configured", node_with_config)
带运行时上下文的节点 #
python
from dataclasses import dataclass
from langgraph.runtime import Runtime
@dataclass
class Context:
db_connection: any
user_id: str
def node_with_runtime(state: State, runtime: Runtime[Context]):
db = runtime.context.db_connection
user_id = runtime.context.user_id
data = db.query(user_id)
return {"data": data}
graph = StateGraph(State, context_schema=Context)
graph.add_node("db_node", node_with_runtime)
异步节点 #
python
async def async_node(state: State):
response = await llm.ainvoke(state["messages"])
return {"messages": [response]}
graph.add_node("async_agent", async_node)
节点返回值 #
节点可以返回以下类型：
python
def node_returns_dict(state: State):
return {"count": state["count"] + 1}
def node_returns_command(state: State):
from langgraph.types import Command
return Command(update={"count": 1}, goto="next")
def node_returns_none(state: State):
return None
添加节点的方式 #
python
def my_node(state: State):
return {"count": 1}
graph.add_node("named_node", my_node)
graph.add_node(my_node)
graph.add_node("lambda_node", lambda state: {"count": 2})
特殊节点 #
START 节点 #
python
from langgraph.graph import START
graph.add_edge(START, "first_node")
graph.set_entry_point("first_node")
END 节点 #
python
from langgraph.graph import END
graph.add_edge("last_node", END)
graph.set_finish_point("last_node")
ToolNode #
LangGraph 提供了预构建的 ToolNode：
python
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
return f"Results for: {query}"
tools = [search]
tool_node = ToolNode(tools)
graph.add_node("tools", tool_node)
节点缓存 #
python
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
import time
def expensive_node(state: State):
time.sleep(2)
return {"result": state["input"] * 2}
graph.add_node("expensive", expensive_node, cache_policy=CachePolicy(ttl=60))
app = graph.compile(cache=InMemoryCache())
边（Edges） #
边的类型 #
text
┌─────────────────────────────────────────────────────────────┐
│ 边的类型 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 普通边（Normal Edge） │
│ - 固定转换 │
│ - A 总是到 B │
│ │
│ 2. 条件边（Conditional Edge） │
│ - 动态路由 │
│ - 根据状态决定下一个节点 │
│ │
│ 3. 入口点（Entry Point） │
│ - 定义图的起始节点 │
│ │
│ 4. 条件入口点（Conditional Entry Point） │
│ - 动态决定起始节点 │
│ │
└─────────────────────────────────────────────────────────────┘
普通边 #
普通边表示固定的转换：
python
from langgraph.graph import START, END
graph.add_edge(START, "agent")
graph.add_edge("agent", "tools")
graph.add_edge("tools", "agent")
graph.add_edge("agent", END)
条件边 #
条件边根据状态动态路由：
python
def should_continue(state: State):
if state["messages"][-1].tool_calls:
return "tools"
return END
graph.add_conditional_edges(
"agent",
should_continue,
["tools", END]
)
条件边映射 #
使用字典映射返回值到节点名称：
python
def route_by_type(state: State):
return state["type"]
graph.add_conditional_edges(
"router",
route_by_type,
{
"math": "math_agent",
"code": "code_agent",
"general": "general_agent"
}
)
条件边返回多个节点 #
python
def parallel_route(state: State):
return ["worker_1", "worker_2", "worker_3"]
graph.add_conditional_edges("splitter", parallel_route)
入口点 #
python
from langgraph.graph import START
graph.add_edge(START, "first_node")
graph.set_entry_point("first_node")
条件入口点 #
python
from langgraph.graph import START
def route_entry(state: State):
if state.get("priority") == "high":
return "priority_handler"
return "normal_handler"
graph.add_conditional_edges(START, route_entry)
控制流模式 #
顺序执行 #
text
START ───> Node A ───> Node B ───> Node C ───> END
python
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", "c")
graph.add_edge("c", END)
循环执行 #
text
┌──────────────────────┐
│ │
▼ │
START ───> Agent ───> Tools ────┘
│
│ (完成)
▼
END
python
def should_continue(state: State):
if state["messages"][-1].tool_calls:
return "tools"
return END
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")
分支执行 #
text
┌───> Node B ───┐
│ │
START ───> Router ──┼───> Node C ───┼───> Merge ───> END
│ │
└───> Node D ───┘
python
def route(state: State):
return state["route_to"]
graph.add_edge(START, "router")
graph.add_conditional_edges("router", route, ["b", "c", "d"])
graph.add_edge("b", "merge")
graph.add_edge("c", "merge")
graph.add_edge("d", "merge")
graph.add_edge("merge", END)
并行执行 #
text
┌───> Worker 1 ───┐
│ │
START ───> Split ───┼───> Worker 2 ───┼───> Combine ───> END
│ │
└───> Worker 3 ───┘
python
from langgraph.types import Send
def split_work(state: State):
items = state["items"]
return [Send("worker", {"item": item}) for item in items]
graph.add_edge(START, "splitter")
graph.add_conditional_edges("splitter", split_work)
graph.add_edge("worker", "combiner")
graph.add_edge("combiner", END)
Send 对象 #
Send 用于动态创建边，实现 Map-Reduce 等模式。
基本用法 #
python
from langgraph.types import Send
def generate_tasks(state: State):
topics = state["topics"]
return [
Send("process", {"topic": topic})
for topic in topics
]
graph.add_conditional_edges("generator", generate_tasks)
Map-Reduce 示例 #
python
from typing import TypedDict, Annotated
from operator import add
from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
topics: list[str]
jokes: Annotated[list[str], add]
def generate_topics(state: State):
return {"topics": ["cats", "dogs", "birds"]}
def map_jokes(state: State):
return [Send("generate_joke", {"topic": t}) for t in state["topics"]]
def generate_joke(state: State):
topic = state["topic"]
joke = llm.invoke(f"Tell a joke about {topic}")
return {"jokes": [joke]}
def combine(state: State):
return {"result": "\n".join(state["jokes"])}
graph = StateGraph(State)
graph.add_node("generate_topics", generate_topics)
graph.add_node("generate_joke", generate_joke)
graph.add_node("combine", combine)
graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", map_jokes)
graph.add_edge("generate_joke", "combine")
graph.add_edge("combine", END)
Command 对象 #
Command 是一个多功能原语，用于控制图执行。
基本用法 #
python
from langgraph.types import Command
from typing import Literal
def my_node(state: State) -> Command[Literal["next_node"]]:
return Command(
update={"count": state["count"] + 1},
goto="next_node"
)
条件跳转 #
python
def router(state: State) -> Command:
if state["type"] == "math":
return Command(goto="math_agent")
elif state["type"] == "code":
return Command(goto="code_agent")
return Command(goto="general_agent")
子图导航 #
python
def subgraph_node(state: State) -> Command:
return Command(
update={"result": "done"},
goto="parent_node",
graph=Command.PARENT
)
中断恢复 #
python
from langgraph.types import interrupt
def human_review(state: State):
answer = interrupt("Please review:")
return {"approved": answer}
result = graph.invoke({"messages": [...]}, config)
result = graph.invoke(Command(resume="approved"), config)
边的最佳实践 #
1. 清晰的路由函数命名 #
python
def should_use_tools(state: State):
"""判断是否需要使用工具"""
return "tools" if state["messages"][-1].tool_calls else END
def route_by_intent(state: State):
"""根据意图路由到不同的处理器"""
intent = state["intent"]
return INTENT_TO_NODE.get(intent, "general")
2. 使用类型注解 #
python
from typing import Literal
def route(state: State) -> Literal["tools", END]:
if state["messages"][-1].tool_calls:
return "tools"
return END
3. 避免复杂的条件逻辑 #
python
def bad_route(state: State):
if state["a"] and state["b"] or state["c"]:
if state["d"]:
return "x"
else:
return "y"
return "z"
def good_route(state: State):
if should_do_x(state):
return "x"
if should_do_y(state):
return "y"
return "z"
4. 使用常量定义节点名 #
python
NODE_AGENT = "agent"
NODE_TOOLS = "tools"
NODE_REVIEW = "review"
graph.add_node(NODE_AGENT, agent_node)
graph.add_node(NODE_TOOLS, tool_node)
graph.add_edge(NODE_AGENT, NODE_TOOLS)
调试节点和边 #
打印执行路径 #
python
for step in app.stream({"messages": [...]}):
for node_name, output in step.items():
print(f"Executed: {node_name}")
print(f"Output: {output}")
可视化图结构 #
python
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))
获取 Mermaid 图 #
python
print(app.get_graph().draw_mermaid())
常见模式 #
Agent 循环模式 #
python
def agent_node(state: State):
return {"messages": [llm.invoke(state["messages"])]}
def should_continue(state: State):
if state["messages"][-1].tool_calls:
return "tools"
return END
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")
多 Agent 路由模式 #
python
def supervisor(state: State):
response = llm.invoke(state["messages"])
return {"next_agent": response.next_agent}
def route_to_agent(state: State):
return state["next_agent"]
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_node("writer", writer_node)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
"supervisor",
route_to_agent,
{"researcher": "researcher", "coder": "coder", "writer": "writer"}
)
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
graph.add_edge("writer", END)
错误处理模式 #
python
def risky_operation(state: State):
try:
result = do_something_risky(state)
return {"result": result, "error": None}
except Exception as e:
return {"error": str(e)}
def handle_error(state: State):
if state.get("error"):
return "error_handler"
return "continue"
graph.add_node("operation", risky_operation)
graph.add_node("error_handler", error_handler_node)
graph.add_node("continue", continue_node)
graph.add_conditional_edges("operation", handle_error)
graph.add_edge("error_handler", END)
graph.add_edge("continue", END)
下一步 #
现在你已经掌握了节点和边，接下来学习 工具使用，了解如何让 Agent 使用外部工具！
最后更新：2026-03-30