LangGraph Edge 边
本文最后更新于 2026年8月11日
更新中
1.概述
langgraph的边，定义了逻辑如何路由以及流程怎样结束，是智能体不同节点彼此通信的重要组成部分。langgraph有几种类型的边：
- 普通边（Normal Edges） 直接从一个节点进入下一个节点
- 条件边（Conditional Edges） 调用一个函数决定去往哪些节点
- 入口点（Entry Point） 用户输入到达时首先调用的节点
- 条件入口点（Conditional Entry Point） 调用一个函数决定用户输入到达时首先调用哪些节点
一个节点有多个出边，如果一个节点有多个出边，那么所有目标节点都作为下一步并行执行。
2.普通边
langgraph两个节点之间就是一条边，两节点通过add_edge相连形成边，自带头节点START和尾节点END，也可以通过set_entry_point()，graph.set_finish_point()指定头尾节点
langgraph没有下游边的最后节点走完就自动结束，也可以不指向尾节点，但是建议“有始有终”
例：通过add_edge()相连，就是最简单的普通边
from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END
# 状态
class DemoState(TypedDict):
name: str
# 节点
def node1(state: DemoState) -> dict:
return state
# 节点
def node2(state: DemoState) -> dict:
return state
def node3(state: DemoState) -> dict:
return state
def node4(state: DemoState) -> dict:
return state
if __name__ == "__main__":
graph = StateGraph(DemoState)
graph.add_node('node1', node1)
graph.add_node('node2', node2)
graph.add_node('node3', node3)
graph.add_node('node4', node4)
graph.add_edge(START, 'node1')
graph.add_edge('node1', 'node2')
graph.add_edge('node2', 'node3')
graph.add_edge('node3', 'node4')
graph.add_edge('node4', END)
#graph.set_entry_point('node1')
#graph.set_finish_point('node4')
app = graph.compile()
app.get_graph().print_ascii()+-----------+
| __start__ |
+-----------+
*
*
*
+-------+
| node1 |
+-------+
*
*
*
+-------+
| node2 |
+-------+
*
*
*
+-------+
| node3 |
+-------+
*
*
*
+-------+
| node4 |
+-------+
*
*
*
+---------+
| __end__ |
+---------+ 3.条件边
在实际场景中，工作流下个节点可能不是固定的，要根据当前执行状态去确定下一个路由的节点，条件边就可以动态的控制执行的流程，langgraph可以使用路由函数来选择具体要执行的节点。
例：add_conditional_edges()就是添加条件边，使用add_conditional_edges(当前节点,决策函数,路由条件映射)方式调用，将check节点跳到下个节点的决策权交给is_even()方法，根据方法返回值决定跳到even节点还是odd节点
from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END
# 状态
class DemoState(TypedDict):
num: int
# 节点
def check(state: DemoState) -> dict:
return state
def is_even(state: DemoState):
return state['num'] % 2 == 0
# 节点
def even(state: DemoState) -> dict:
print('even')
return state
def odd(state: DemoState) -> dict:
print('odd')
return state
if __name__ == "__main__":
graph = StateGraph(DemoState)
graph.add_node('check', check)
graph.add_node('even', even)
graph.add_node('odd', odd)
graph.add_edge(START, 'check')
graph.add_conditional_edges('check', is_even, {
True: 'even',
False: 'odd'
}
)
graph.add_edge('odd', END)
graph.add_edge('even', END)
app = graph.compile()
app.get_graph().print_ascii() +-----------+
| __start__ |
+-----------+
*
*
*
+-------+
| check |
+-------+
.. ..
. .
. .
+------+ +-----+
| even | | odd |
+------+ +-----+
** **
* *
* *
+---------+
| __end__ |
+---------+ 传值调用
res = app.invoke(DemoState(num=121))
print(res)
res = app.invoke(DemoState(num=12))
print(res)odd
{'num': 121}
even
{'num': 12}条件边构造循环结构
有时，会出现两个节点之间不断循环的情况，例如REACT模式下，大模型调用工具，工具返回后，大模型再次调用工具，直到工具返回正确的结果构成了一个循环，这种场景就可以使用条件边构建条件循环
例：
from typing import TypedDict, Literal
from langgraph.errors import GraphRecursionError
from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END
# 状态
class DemoState(TypedDict):
num: int
total_loop: int
max: int
# 节点
def node_a(state: DemoState) -> dict:
return {
'num': state["num"] + 1,
'total_loop': state['total_loop'] + 1
}
def node_b(state: DemoState) -> dict:
return {
'total_loop': state['total_loop'] + 1
}
def my_loop(state: DemoState) -> Literal['node_b', END]:
if state['total_loop'] >= state['max']:
return END
else:
return 'node_b'
if __name__ == "__main__":
graph = StateGraph(DemoState)
graph.add_node('node_a', node_a)
graph.add_node('node_b', node_b)
graph.add_edge(START, 'node_a')
graph.add_conditional_edges('node_a', my_loop)
graph.add_edge('node_b', 'node_a')
app = graph.compile()
res = app.invoke(input=DemoState(num=0, total_loop=0, max=300))
print(res){'num': 151, 'total_loop': 301, 'max': 300}为防止构成死循环，langgraph支持使用config={'recursion_limit': 200}配置，从底层控制循环次数，这个配置在LangGraph定义节点（Node）就用过
try:
res = app.invoke(
input=DemoState(num=0, total_loop=0, max=300),
config={'recursion_limit': 50}
)
print(res)
except GraphRecursionError as e:
print(e)Recursion limit of 200 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT
Process finished with exit code 0例：模拟ReAct风格智能体
import os
from typing import TypedDict, Annotated
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
from langgraph.constants import START, END
from langgraph.graph import StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from pydantic import Field, BaseModel
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
llm = init_chat_model(
model = 'deepseek-chat',
model_provider = 'openai',
api_key = os.getenv('DSKEY'),
base_url = 'https://api.deepseek.com'
)
class FiledInfo(BaseModel):
"""
定义参数信息
"""
city: str = Field(description='城市')
@tool(args_schema=FiledInfo, description='根据城市名称获取温度')
def tp_tool(city: str):
print('=======tp_tool=======')
if city == '北京':
return 12
elif city == '武汉':
return 23
elif city == '沈阳':
return -10
elif city == '泉州':
return 27
else:
return None
tools = [tp_tool]
llm_with_tool = llm.bind_tools(tools)
SYSTEM_PROMPT = SystemMessage(content='你是一个智能助手，能调用工具和回答问题')
def chat_node(state: AgentState) -> dict:
res = llm_with_tool.invoke( [SYSTEM_PROMPT] + state['messages'] )
return {
'messages': [res]
}
tools_node = ToolNode(tools=tools)
def route_after_chat(state: AgentState):
last_message = state['messages'][-1]
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
return 'tools_node'
return END
if __name__ == '__main__':
graph = StateGraph(AgentState)
graph.add_node('chat_node', chat_node)
graph.add_node('tools_node', tools_node)
graph.add_edge(START, 'chat_node')
graph.add_conditional_edges('chat_node', route_after_chat, ['tools_node', END])
graph.add_edge('tools_node', 'chat_node')
app = graph.compile()
app.get_graph().print_ascii()
resp = app.invoke({"messages": '在吗？沈阳的温度是？'})
#这样也可：resp = app.invoke({"messages": [HumanMessage(content='武汉多少度？')]})
for e in resp['messages']:
print(e)
+-----------+
| __start__ |
+-----------+
*
*
*
+-----------+
| chat_node |
+-----------+
... ...
. .
.. ..
+---------+ +------------+
| __end__ | | tools_node |
+---------+ +------------+
=======tp_tool=======
content='在吗？沈阳的温度是？' additional_kwargs={} response_metadata={} id='69998b70-c1f5-4111-8717-548e8ce5d01a'
content='在的！我来帮你查一下沈阳的温度。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 53, 'prompt_tokens': 296, 'total_tokens': 349, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 40}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': 'bafe5bee-2ac9-45a0-9e16-5b52243bfeb4', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019fa3f5-1393-7142-9d36-f20a157924d2-0' tool_calls=[{'name': 'tp_tool', 'args': {'city': '沈阳'}, 'id': 'call_00_9DxFdFrhliSRF515rmHe7902', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 296, 'output_tokens': 53, 'total_tokens': 349, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}
content='-10' name='tp_tool' id='f53bbf6b-3ad7-40e9-b7c0-3930eb53e58d' tool_call_id='call_00_9DxFdFrhliSRF515rmHe7902'
content='沈阳目前的温度是 **-10℃** 🥶，天气比较冷，出门记得注意保暖哦！' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 23, 'prompt_tokens': 362, 'total_tokens': 385, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 106}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': 'dbf2d398-acbe-41e0-bee8-9756f932b649', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019fa3f5-18a9-7b92-a933-a61772087c40-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 362, 'output_tokens': 23, 'total_tokens': 385, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}
条件边构造条件入口点
前面例子的添加第一个节点add_edge(START, )或set_entry_point()，就是工作流的入口点，有时需要根据输入动态判断走某个入口，一个流程设置多个入口按需进入，这就是条件入口点，用户输入抵达，执行函数判断进入哪个入口点。
langgraph同样使用条件边add_conditional_edges(START,决策函数,路由条件映射)实现条件入口点，使用方式和普通条件边完全相同。
例：
from typing import TypedDict, Literal
from langgraph.constants import START, END
from langgraph.graph import StateGraph
class DemoState(TypedDict):
num: int
def node1(state: DemoState):
print('node1')
return state
def node2(state: DemoState):
print('node2')
return state
def node3(state: DemoState):
print('node3')
return state
def check(state: DemoState) -> Literal['n1', 'n2', 'n3', END]:
if state['num'] == 1:
return 'n1'
if state['num'] == 2:
return 'n2'
if state['num'] == 3:
return 'n3'
else:
return END
if __name__ == '__main__':
graph = StateGraph(DemoState)
graph.add_node('node1', node1)
graph.add_node('node2', node2)
graph.add_node('node3', node3)
graph.add_conditional_edges(START, check, {
'n1': 'node1',
'n2': 'node2',
'n3': 'node3',
})
graph.add_edge('node1', END)
graph.add_edge('node2', END)
graph.add_edge('node3', END)
app = graph.compile()
app.get_graph().print_ascii()
res = app.invoke( DemoState(num=2) )
print(res) +-----------+
| __start__ |
+-----------+..
... . ...
... . ...
.. . ..
+-------+ +-------+ +-------+
| node1 |* | node2 | | node3 |
+-------+ *** +-------+ **+-------+
*** * ***
*** * ***
** * **
+---------+
| __end__ |
+---------+
node2
{'num': 2}"如果文章对您有帮助，可以请作者喝杯咖啡吗？"
微信支付
支付宝
LangGraph Edge 边
https://blog.liuzijian.com/post/2026/07/05/langgraph-edge/