跑通一个最小状态图)
学习目标:
- 跑通一个最小状态图。
- 理解
State / Node / Edge的关系。 - 观察条件边如何改变执行路径。
- StateGraph
StateGraph 是 LangGraph 的核心概念,它用"图"的方式来编排任务执行流程。
StateGraph 的三个核心概念
— State(状态):数据在节点之间流动的容器,由 TypedDict 定义结构。
— Node(节点):图中的执行单元,每个节点读取并修改状态。
— Edge(边):定义节点之间的连接关系。
— Conditional Edge(条件边):根据当前状态选择走哪条路径。
注意事项:
1. 节点拿到的是“当前完整状态”,LangGraph 的 StateGraph 本质就是：节点读取共享状态，然后返回 Partial<State>，也就是状态的一部分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_tool=any(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_tool=True 时才会走到此节点）。 这个节点展示"需要工具"的分支逻辑。在真实场景中,这里会调用 搜索引擎或 API,本 demo 出于学习目的,只是模拟生成一个检索结果。 注意此节点的执行路径可能被跳过:如果 analyze_question 认为 不需要工具,条件边会直接绕开此节点。""" print(f"[retrieve_info] 模拟检索:{state['question']}")result=f'[检索结果] 关于"{state["question"]}"的参考资料'print(f"[retrieve_info] 检索到:{result}")return{"answer":result,"steps":state["steps"]+["retrieve_info"],}def generate_answer(state: GraphState)->dict:"""根据是否有检索结果,以不同策略合成最终回答。 这个节点展示了状态驱动的分支行为——它的执行逻辑依赖上游节点 在状态中留下的数据。如果有检索结果,就做综合回答；否则直接回答。 条件边决定节点是否执行,而节点内部还可以根据状态做进一步的逻辑判断。"""ifstate["answer"]: answer=f"综合检索结果回答:{state['answer']}"else: answer=f"直接回答:{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 分支")return"retrieve"print("[route_decision] 不需要工具 → 走 direct 分支")return"direct"def build_graph()->StateGraph:"""构建并编译状态图。 建图流程分为四步:1. 创建 StateGraph 实例,传入状态结构定义。2. 用 add_node 注册所有节点（名字→函数的映射）。3. 用 set_entry_point / add_edge / add_conditional_edges 定义边。4. 调用 compile()编译为可执行的图。 注意编译前的图是"定义态",compile()之后才是"运行态"。 编译过程会做拓扑检查,确保所有节点可达、没有死环。""" builder=StateGraph(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}")if"steps"instate_update: all_steps=state_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()graph=build_graph()# 测试用例 1:不包含关键词 → need_tool=False → direct 分支run_case(graph,"Python 的列表推导式怎么用？")# 测试用例 2:包含关键词 → need_tool=True → 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}")if"steps"instate_update: all_steps=state_update["steps"]print()重点讲解
1. graph.stream(initial_state)是什么 它会启动整张图的执行，并且不是等整张图跑完才一次性返回结果，而是： 跑完一个节点 立刻产出一个事件 event 再继续下一个节点 所以它是“流式”的。2. 和它对应的是 graph.invoke(initial_state)： invoke()：只给你最后结果 stream()：把中间每一步也给你看 注意事项: event 不是完整状态。 state_update 只是这个节点“这一次返回的更新内容”。
