Nodes & Edges:节点与边

面试官:「LangGraph 里 Node 和 Edge 是什么关系?」

你:「节点是处理函数,边是它们之间的流向。」

面试官:「怎么实现条件分支?」

你:「用条件边,返回目标节点名来决定下一步。」

🤔 这一章,把节点和边彻底搞懂。

3.1 节点类型详解

LangGraph 支持多种类型的节点,每种都有其适用场景。

3.1.1 函数节点

python
# 最基础的节点类型:普通 Python 函数 def process_input(state: State) -> dict: data = state["input"] return {"processed": data.upper()} graph.add_node("process", process_input)

3.1.2 LLM 节点

python
from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4") # LLM 作为节点:直接返回消息 def llm_node(state: State) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]}

3.1.3 工具节点

python
from langchain_core.tools import tool @tool def search_database(query: str) -> str: """搜索数据库""" return f"查询结果: {query}" @tool def send_email(to: str, content: str) -> str: """发送邮件""" return f"已发送给 {to}" tools = [search_database, send_email] llm_with_tools = llm.bind_tools(tools) def tool_node(state: State) -> dict: # LLM 决定调用哪个工具 last_msg = state["messages"][-1] if last_msg.tool_calls: tool_call = last_msg.tool_calls[0] result = tools[tool_call.name].invoke(tool_call.args) return {"messages": [{"role": "tool", "content": result}]} return {"messages": []}

3.2 边的进阶用法

3.2.1 普通边(Fixed Edge)

python
# 固定从 A 到 B graph.add_edge("node_a", "node_b")

3.2.2 条件边(Conditional Edge)

python
from typing import Literal def route_after_llm(state: State) -> Literal["search", "calculate", "END"]: last_msg = state["messages"][-1].content if "搜索" in last_msg or "查询" in last_msg: return "search" elif any(kw in last_msg for kw in ["计算", "多少", "数字"]): return "calculate" else: return "END" # 条件边:从 llm_node 根据返回值决定下一个节点 graph.add_conditional_edges( "llm_node", route_after_llm, { "search": "search_node", "calculate": "calculate_node", "END": END } )
图 3-1:条件边路由示意
llm_node search calculate END 搜索关键词 计算关键词 其他

3.3 节点执行详解

每个节点接收 当前 State,处理后返回部分更新,由 Reducer 合并到 State 中。

图 3-2:节点执行流程
当前 State {"input": "hello"} process_node() return {"out": "HELLO"} Reducer 合并 {"input":"hello", "out":"HELLO"} step 1: 读取 step 2: 处理 step 3: 写回
关键点

节点返回的是部分更新,不是完整 State。Reducer 负责合并多个节点的更新(第 2 章内容)。

3.4 完整工作流示例

python
# agents.py - Agent 定义 from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: Annotated[list, add] # ... def create_agent() -> StateGraph: graph = StateGraph(AgentState) # 添加节点 graph.add_node("llm", llm_node) graph.add_node("tools", tool_node) # 添加边 graph.add_edge(START, "llm") graph.add_conditional_edges("llm", route_to_tools, {...}) graph.add_edge("tools", "llm") # 循环回到 LLM return graph.compile()
图 3-3:聊天 Agent 完整工作流
START llm_node tool_node END 外部工具 需要工具 直接回答 →END 对话结束

3.5 本章小结

  • 节点类型:函数节点、LLM 节点、工具节点
  • 边类型:普通边(固定)、条件边(动态路由)
  • add_conditional_edges:根据函数返回值决定下一个节点
  • 合理组织代码结构:分离 Agent 定义和调用逻辑
下章预告

第 4 章我们将学习 条件分支与路由:复杂的条件逻辑、循环迭代、以及如何实现 ReAct 模式。