学习目标:跑通一个最小状态图。理解State / Node / Edge的关系。观察条件边如何改变执行路径。StateGraphStateGraph 是 LangGraph 的核心概念,它用图的方式来编排任务执行流程。StateGraph 的三个核心概念— State(状态):数据在节点之间流动的容器,由 TypedDict 定义结构。— Node(节点):图中的执行单元,每个节点读取并修改状态。— Edge(边):定义节点之间的连接关系。— Conditional Edge(条件边):根据当前状态选择走哪条路径。注意事项:1. 节点拿到的是“当前完整状态”,LangGraph 的 StateGraph 本质就是节点读取共享状态然后返回 PartialState也就是状态的一部分2. 节点不需要返回完整 state只返回要修改的字段3. 下一个节点看到的是“合并后的状态”建图流程分为四步:1. 创建 StateGraph 实例,传入状态结构定义。2. 用 add_node 注册所有节点名字→函数的映射。3. 用 set_entry_point / add_edge / add_conditional_edges 定义边。4. 调用 compile() 编译为可执行的图。Day1:StateGraph 入门学习实现。 StateGraph 是 LangGraph 的核心概念,它用图的方式来编排任务执行流程。 本 demo 通过一个最小状态图不接 LLM,帮助你理解三个核心概念: — State状态:数据在节点之间流动的容器,由 TypedDict 定义结构。 — Node节点:图中的执行单元,每个节点读取并修改状态。 — Edge边:定义节点之间的连接关系。 — Conditional Edge条件边:根据当前状态选择走哪条路径。 运行本 demo 会执行两个测试用例,展示条件边如何让同一张图走出两种不同的路径。 from typingimportLiteral, TypedDict from langgraph.graphimportEND, StateGraph class GraphState(TypedDict):最小状态定义,控制在4个字段以内。 每个字段承担不同的角色: — question:外部输入,由图调用者传入,节点只读取不修改。 — need_tool:节点分析后产出的布尔标记,驱动条件边做路由决策。 — answer:多个节点接力构建的输出字段,展示状态如何逐节点累积。 — steps:路径追踪列表,每个节点将自己的名字追加到末尾, 最终形成完整的执行路径,用于验证图的执行顺序。 question: str need_tool: bool answer: str steps: list[str]def analyze_question(state: GraphState)-dict:第一个节点:分析问题是否包含需要工具检索的关键词。 这是图中的入口节点。它读取 question 并判断 need_tool, 后续的条件边将根据这个标记决定走哪条分支。 注意:节点函数接收当前完整状态,返回一个字典只包含要更新的字段。 StateGraph 会把返回的字典合并到状态中——不需要返回所有字段。 need_tool_words[搜索,查找,计算,翻译]need_toolany(wordinstate[question]forwordinneed_tool_words)print(f[analyze_question] 问题:{state[question]})print(f[analyze_question] 判断结果:need_tool{need_tool})return{need_tool:need_tool,steps:state[steps][analyze_question],}def retrieve_info(state: GraphState)-dict:模拟工具检索仅当need_toolTrue 时才会走到此节点。 这个节点展示需要工具的分支逻辑。在真实场景中,这里会调用 搜索引擎或 API,本 demo 出于学习目的,只是模拟生成一个检索结果。 注意此节点的执行路径可能被跳过:如果 analyze_question 认为 不需要工具,条件边会直接绕开此节点。 print(f[retrieve_info] 模拟检索:{state[question]})resultf[检索结果] 关于{state[question]}的参考资料print(f[retrieve_info] 检索到:{result})return{answer:result,steps:state[steps][retrieve_info],}def generate_answer(state: GraphState)-dict:根据是否有检索结果,以不同策略合成最终回答。 这个节点展示了状态驱动的分支行为——它的执行逻辑依赖上游节点 在状态中留下的数据。如果有检索结果,就做综合回答否则直接回答。 条件边决定节点是否执行,而节点内部还可以根据状态做进一步的逻辑判断。ifstate[answer]: answerf综合检索结果回答:{state[answer]}else: answerf直接回答:{state[question]} 是一个基础问题,直接回答即可。print(f[generate_answer] 生成回答:{answer})return{answer:answer,steps:state[steps][generate_answer],}def finalize(state: GraphState)-dict:终结点:输出最终结果快照,然后结束。 这个节点之后不再有处理逻辑,直接连接到 END。 它主要负责输出最终状态供学习者观察完整的执行结果。 print(f[finalize] 最终 answer {state[answer]})print(f[finalize] 执行路径 steps {state[steps]})return{steps:state[steps][finalize],}def route_decision(state: GraphState)-Literal[retrieve,direct]:条件边路由函数:根据 need_tool 选择下一个节点。 条件边是 StateGraph 区别于普通函数链的关键特性: 普通函数链是固定的 A→B→C,而条件边让执行路径在运行时动态决定。 路由函数返回一个字符串键,图会查找对应的边映射表来确定下一个节点。 这种函数 映射表的模式使得路由逻辑既可读又灵活。ifstate.get(need_tool, False): print([route_decision] 需要工具 → 走 retrieve 分支)returnretrieveprint([route_decision] 不需要工具 → 走 direct 分支)returndirectdef build_graph()-StateGraph:构建并编译状态图。 建图流程分为四步:1. 创建 StateGraph 实例,传入状态结构定义。2. 用 add_node 注册所有节点名字→函数的映射。3. 用 set_entry_point / add_edge / add_conditional_edges 定义边。4. 调用 compile()编译为可执行的图。 注意编译前的图是定义态,compile()之后才是运行态。 编译过程会做拓扑检查,确保所有节点可达、没有死环。 builderStateGraph(GraphState)# 注册 4 个节点builder.add_node(analyze_question, analyze_question)builder.add_node(retrieve_info, retrieve_info)builder.add_node(generate_answer, generate_answer)builder.add_node(finalize, finalize)# 设置图的入口点builder.set_entry_point(analyze_question)# 条件边:从 analyze_question 出发,根据路由函数决定走向builder.add_conditional_edges(analyze_question, route_decision,{retrieve:retrieve_info,direct:generate_answer},)# 普通边:确定性的节点间连接builder.add_edge(retrieve_info,generate_answer)builder.add_edge(generate_answer,finalize)# builder.set_finish_point(finalize)returnbuilder.compile()def run_case(graph: StateGraph, question: str)-None:运行单个测试用例,展示节点执行顺序和状态变化。 graph.stream()是 LangGraph 的流式执行方法,逐节点产出事件。 每个事件是一个 dict:{节点名: 该节点产出的状态更新}。 这样可以在每个节点执行后立即看到它对状态做了哪些修改。 相比 graph.invoke()只返回最终状态,stream()更适合学习目的, 因为它暴露了图执行的中间过程。 print(f 输入:{question})initial_state: GraphState{question:question,need_tool:False,answer:,steps:[],}all_steps: list[str][]foreventingraph.stream(initial_state):fornode_name, state_updateinevent.items(): print(f--- 节点 {node_name} 产出的状态更新 ---)forkey, valueinstate_update.items(): print(f {key} {value})ifstepsinstate_update: all_stepsstate_update[steps]print()print(f执行路径:{ → .join(all_steps)})print(*60)def main()-None:演示条件边在不同输入下的两种执行路径。 测试用例1不包含关键词走 direct 分支:3 个节点执行。 测试用例2包含搜索关键词走 retrieve 分支:4 个节点执行。 同一张图,不同的输入,走出了不同的路径——这就是条件边的价值。 print(*60)print( Day 1:StateGraph 入门)print( 条件边演示:同一图结构,两种执行路径)print(*60)print()graphbuild_graph()# 测试用例 1:不包含关键词 → need_toolFalse → direct 分支run_case(graph,Python 的列表推导式怎么用)# 测试用例 2:包含关键词 → need_toolTrue → retrieve 分支# run_case(graph, 搜索一下 Python 语法教程)if__name____main__:main()其中foreventingraph.stream(initial_state):fornode_name, state_updateinevent.items(): print(f--- 节点 {node_name} 产出的状态更新 ---)forkey, valueinstate_update.items(): print(f {key} {value})ifstepsinstate_update: all_stepsstate_update[steps]print()重点讲解1. graph.stream(initial_state)是什么 它会启动整张图的执行并且不是等整张图跑完才一次性返回结果而是 跑完一个节点 立刻产出一个事件 event 再继续下一个节点 所以它是“流式”的。2. 和它对应的是 graph.invoke(initial_state) invoke()只给你最后结果 stream()把中间每一步也给你看 注意事项: event 不是完整状态。 state_update 只是这个节点“这一次返回的更新内容”。