LangGraph核心概念详解，节点、边、状态、条件边

LangGraph核心概念详解节点、边、状态、条件边上一篇用一个例子展示了LangGraph的能力。这一篇我们把核心概念拆开讲清楚。LangGraph的四个核心概念是节点、边、状态、条件边。理解了这四个LangGraph就入门了。状态State状态是LangGraph最核心的概念。整个图的执行过程中所有节点共享一个状态对象。状态通常用TypedDict来定义。每个字段代表一种信息。fromtypingimportTypedDict,List,OptionalclassAgentState(TypedDict):messages:List[str]# 对话历史current_task:str# 当前任务results:List[str]# 中间结果retry_count:int# 重试次数status:str# 当前状态每个节点函数接收状态作为参数返回一个字典。返回的字典会合并到当前状态里。defsearch_node(state:AgentState)-dict:# 从状态里读取信息taskstate[current_task]# 执行搜索resultf搜索{task}的结果# 返回要更新的字段return{results:state[results][result],status:searched,}这里有个重要的设计。节点返回的不是完整的新状态只是要更新的字段。LangGraph会自动把返回值合并到状态里。没返回的字段保持不变。这种设计让节点之间的耦合度很低。每个节点只关心自己需要读什么、写什么不用管其他节点。节点Node节点是图里的处理步骤。每个节点是一个函数接收状态返回状态更新。defanalyze_node(state:AgentState)-dict:resultsstate[results]# 分析结果analysisf分析了{len(results)}条结果return{messages:state[messages][analysis],status:analyzed,}节点的函数签名是固定的。参数是当前状态返回值是状态更新字典。添加节点用add_node方法。第一个参数是节点名字第二个参数是节点函数。fromlanggraph.graphimportStateGraph workflowStateGraph(AgentState)workflow.add_node(search,search_node)workflow.add_node(analyze,analyze_node)workflow.add_node(report,report_node)节点名字是字符串标识后面添加边的时候用这个名字来引用。边Edge边定义了节点之间的跳转关系。从节点A执行完后走向节点B。# search执行完后走向analyzeworkflow.add_edge(search,analyze)# analyze执行完后走向reportworkflow.add_edge(analyze,report)普通的边是无条件的。A执行完一定走向B。还有一个特殊的边入口边。用set_entry_point来设置图的起始节点。# 图从search节点开始workflow.set_entry_point(search)还有一个特殊的终止节点END。走到END就表示图执行结束。# report执行完后结束fromlanggraph.graphimportEND workflow.add_edge(report,END)条件边Conditional Edge条件边是LangGraph最强大的功能。它允许根据当前状态动态决定下一步走向哪个节点。defdecide_next(state:AgentState)-str:statusstate[status]ifstatusanalyzed:returnreportelifstatusneed_more_data:returnsearchelse:returnerror_handler# 添加条件边workflow.add_conditional_edges(analyze,# 从哪个节点出发decide_next,# 判断函数)条件边的判断函数接收状态返回一个字符串。这个字符串就是下一步要走向的节点名字。判断函数里可以写任意复杂的逻辑。读状态、做判断、返回目标节点。条件边是实现分支和循环的关键。分支就是条件边返回不同的节点名字。循环就是条件边返回之前的节点。一个完整的例子把四个概念串起来做一个完整的Agent工作流。fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,List# 1. 定义状态classWorkState(TypedDict):question:strsearch_results:List[str]answer:strneeds_search:boolquality_ok:boolretry_count:int# 2. 定义节点defdecide_node(state:WorkState)-dict:判断是否需要搜索questionstate[question]# 简单问题不需要搜索iflen(question)10:return{needs_search:False}return{needs_search:True}defsearch_node(state:WorkState)-dict:搜索资料questionstate[question]resultf关于{question}的搜索结果resultsstate.get(search_results,[])[result]return{search_results:results}defanswer_node(state:WorkState)-dict:生成回答questionstate[question]resultsstate.get(search_results,[])ifresults:answerf根据{len(results)}条资料回答{question}的答案是...else:answerf直接回答{question}的答案是...return{answer:answer}defcheck_node(state:WorkState)-dict:检查回答质量answerstate[answer]retrystate.get(retry_count,0)# 模拟质量检查iflen(answer)15orretry2:return{quality_ok:True}return{quality_ok:False,retry_count:retry1}# 3. 定义条件判断函数defroute_after_decide(state:WorkState)-str:ifstate[needs_search]:returnsearchreturnanswerdefroute_after_check(state:WorkState)-str:ifstate[quality_ok]:returnENDreturnanswer# 质量不行重新回答# 4. 构建图workflowStateGraph(WorkState)workflow.add_node(decide,decide_node)workflow.add_node(search,search_node)workflow.add_node(answer,answer_node)workflow.add_node(check,check_node)workflow.set_entry_point(decide)workflow.add_conditional_edges(decide,route_after_decide)workflow.add_edge(search,answer)workflow.add_edge(answer,check)workflow.add_conditional_edges(check,route_after_check)# 5. 编译并运行appworkflow.compile()resultapp.invoke({question:什么是RAG检索增强生成技术,search_results:[],answer:,needs_search:False,quality_ok:False,retry_count:0,})print(f回答:{result[answer]})print(f搜索次数:{len(result.get(search_results,[]))})print(f重试次数:{result.get(retry_count,0)})这个工作流的流程是。先判断要不要搜索要搜就搜搜完生成回答。不要搜就直接生成回答。生成后检查质量不行就重来。编译和运行图构建好了以后需要编译才能运行。编译会做一些优化和验证。appworkflow.compile()运行有两种方式。invoke是同步运行等全部执行完返回最终状态。resultapp.invoke(initial_state)stream是流式运行每执行一个节点就返回一次中间状态。适合前端展示执行进度。foreventinapp.stream(initial_state):print(event)几个实用技巧限制循环次数。条件边可能导致无限循环。在状态里加一个计数器超过次数就强制结束。defroute_after_check(state:WorkState)-str:ifstate[quality_ok]orstate.get(retry_count,0)3:returnENDreturnanswer用Annotated减少代码。如果状态字段是列表每次都要手动拼接。用Annotated和reducer可以简化。fromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassState(TypedDict):messages:Annotated[list,add_messages]# 自动追加而不是覆盖保存中间状态。用stream可以看到每一步的中间状态方便调试。下一篇讲状态管理实战。在多轮交互中怎么维护Agent的状态怎么处理复杂的对话上下文。