企业官网建设流程全解析

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:# 从状态里读取信息task=state["current_task"]# 执行搜索result=f"搜索{task}的结果"# 返回要更新的字段return{"results":state["results"]+[result],"status":"searched",}

这里有个重要的设计。节点返回的不是完整的新状态，只是要更新的字段。LangGraph会自动把返回值合并到状态里。没返回的字段保持不变。

这种设计让节点之间的耦合度很低。每个节点只关心自己需要读什么、写什么，不用管其他节点。

节点Node

节点是图里的处理步骤。每个节点是一个函数，接收状态，返回状态更新。

defanalyze_node(state:AgentState)->dict:results=state["results"]# 分析结果analysis=f"分析了{len(results)}条结果"return{"messages":state["messages"]+[analysis],"status":"analyzed",}

节点的函数签名是固定的。参数是当前状态，返回值是状态更新字典。

添加节点用add_node方法。第一个参数是节点名字，第二个参数是节点函数。

fromlanggraph.graphimportStateGraph workflow=StateGraph(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:status=state["status"]ifstatus=="analyzed":return"report"elifstatus=="need_more_data":return"search"else:return"error_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:"""判断是否需要搜索"""question=state["question"]# 简单问题不需要搜索iflen(question)<10:return{"needs_search":False}return{"needs_search":True}defsearch_node(state:WorkState)->dict:"""搜索资料"""question=state["question"]result=f"关于'{question}'的搜索结果"results=state.get("search_results",[])+[result]return{"search_results":results}defanswer_node(state:WorkState)->dict:"""生成回答"""question=state["question"]results=state.get("search_results",[])ifresults:answer=f"根据{len(results)}条资料回答：{question}的答案是..."else:answer=f"直接回答：{question}的答案是..."return{"answer":answer}defcheck_node(state:WorkState)->dict:"""检查回答质量"""answer=state["answer"]retry=state.get("retry_count",0)# 模拟质量检查iflen(answer)>15orretry>=2:return{"quality_ok":True}return{"quality_ok":False,"retry_count":retry+1}# 3. 定义条件判断函数defroute_after_decide(state:WorkState)->str:ifstate["needs_search"]:return"search"return"answer"defroute_after_check(state:WorkState)->str:ifstate["quality_ok"]:returnENDreturn"answer"# 质量不行，重新回答# 4. 构建图workflow=StateGraph(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. 编译并运行app=workflow.compile()result=app.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)}")

这个工作流的流程是。先判断要不要搜索，要搜就搜，搜完生成回答。不要搜就直接生成回答。生成后检查质量，不行就重来。

编译和运行

图构建好了以后，需要编译才能运行。编译会做一些优化和验证。

app=workflow.compile()

运行有两种方式。

invoke是同步运行，等全部执行完返回最终状态。

result=app.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:returnENDreturn"answer"

用Annotated减少代码。如果状态字段是列表，每次都要手动拼接。用Annotated和reducer可以简化。

fromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassState(TypedDict):messages:Annotated[list,add_messages]# 自动追加而不是覆盖

保存中间状态。用stream可以看到每一步的中间状态，方便调试。

下一篇讲状态管理实战。在多轮交互中怎么维护Agent的状态，怎么处理复杂的对话上下文。