LangGraph入门很简陋的入门状态第一部分LangGraph 本质上是一个低层的 Agent 编排与运行框架重点不是“帮你写 Prompt”而是管理状态、流程、持久化、流式输出和人工介入模型 → 工具 → 模型 → 工具 → 模型 → END,实际上是一个循环1. 五个核心概念1.1 State: 共享状态State可以理解为整个 Agent 共用的一张工作表例如{question:什么是 RAG,intent:技术问题,documents:[文档1,文档2],answer:RAG 是……,messages:[...]}所有节点都能读取它但是节点一般不应该直接修改原state而是返回{answer:新的答案}意思是我要更新state中的answer字段官方对节点的定义就是State → Partial也就是接收完整状态 → 返回部分状态更新1.2 Node节点Node 本质上就是一个普通 Python 函数defnode_name(state):# 1. 读取 State# 2. 执行任务# 3. 返回需要更新的字段return{字段:新值}例如defclassify_question(state):questionstate[question]ifPythoninquestion:intentpythonelse:intentgeneralreturn{intent:intent}1.3 Edge边Edge决定一个节点执行完接下来执行谁普通边 顺序就是 执行节点A ↓ 节点A执行结束 ↓ 沿着Edge前往节点B ↓ 执行节点B builder.add_edge(节点A,节点B)条件边 执行分类节点 ↓ 分类节点执行结束 ↓ 调用路由函数 ↓ 根据路由函数返回值 ↓ 选择下一个节点 builder.add_conditional_edges( 分类节点, 路由函数 )节点负责“做事情”Edge 负责“做完以后去哪”。普通 Edge 固定去一个节点条件 Edge 先通过路由函数判断再决定去哪。1.4 START / ENDSTART和END是两个虚拟节点START图从哪里开始END图在哪里结束builder.add_edge(START,classify)builder.add_edge(answer,END)1.5 CheckpointerCheckpointer 可以理解为Agent 的自动存档系统每执行到一个关键节点就保存当前 State 是什么下一步准备执行谁属于哪个 thread有哪些 interrupt它让 Agent 可以多轮对话中断后继续失败后恢复查看历史状态时间旅行2. 第一个 LangGraph线性图先用普通 Python 函数理解图安装pipinstall-Ulanggraph langchain当前官方 Python 文档要求 Python 3.10 或更高版本。完整代码fromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,END# 1. 定义 StateclassAgentState(TypedDict):question:str#输入问题normalized_question:NotRequired[str]#标准化后的问题answer:NotRequired[str]#答案# State是整个图的共享数据结构所有节点都通过它读写数据# 2. 定义节点每个节点是一个纯函数接收 state返回要更新的字段defnormalize_question(state:AgentState):print(执行 normalize_question)questionstate[question]normalizedquestion.strip()# 只返回要更新的字段return{normalized_question:normalized}defgenerate_answer(state:AgentState):print(执行 generate_answer)questionstate[normalized_question]return{answer:f已经收到你的问题{question}}# 3. 创建图builderStateGraph(AgentState)# 4. 添加节点builder.add_node(normalize,normalize_question)builder.add_node(answer,generate_answer)# 5. 添加边builder.add_edge(START,normalize)builder.add_edge(normalize,answer)builder.add_edge(answer,END)# 6. 编译graphbuilder.compile()# 7. 调用图resultgraph.invoke({question: 什么是 RAG })print(result)3. 为什么节点只返回“要更新的字段错误的思路defnode(state):state[answer]结果returnstate推荐写法defnode(state):return{answer:结果}因为LangGraph 要自己管理状态合并并行更新Checkpoint恢复执行状态历史Reducer如果到处直接修改原对象状态变化会越来越难追踪。4. 条件分支add_conditional_edges​ → Python回答 →START → 问题分类 END​ → 普通回答 →代码这段代码看懂就应该看懂了(废话哈哈)fromtypingimportLiteralfromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,ENDclassAgentState(TypedDict):question:str# 输入问题route:NotRequired[str]# 路由决策python 或 generalanswer:NotRequired[str]# 答案# 问题分类defclassify_question(state:AgentState):questionstate[question].lower()# 转为小写#根据问题是否包含 python 来决定路由ifpythoninquestion:routepythonelse:routegeneralreturn{route:route}# 路由函数根据路由选择对应的路由函数注意不是节点。defroute_question(state:AgentState)-Literal[python_answer,general_answer]:ifstate[route]python:returnpython_answerreturngeneral_answerdefpython_answer(state:AgentState):return{answer:f这是一个 Python 问题{state[question]}}defgeneral_answer(state:AgentState):return{answer:f这是一个普通问题{state[question]}}builderStateGraph(AgentState)builder.add_node(classify,classify_question)builder.add_node(python_answer,python_answer)builder.add_node(general_answer,general_answer)builder.add_edge(START,classify)builder.add_conditional_edges(classify,route_question# 路由函数用这个函数来选择路由)builder.add_edge(python_answer,END)builder.add_edge(general_answer,END)graphbuilder.compile()resultgraph.invoke({# question: Python 的装饰器是什么question:明天天气怎么样})print(result[answer])5. Reducer(必须掌握点…(gpt说的))Reducer 是状态合并策略决定多个值怎么合并到一起。假设 State 里有{logs:[节点A执行完成]}节点 B 返回return{logs:[节点B执行完成]}默认情况下新值会覆盖旧值{logs:[节点B执行完成]}但我们想要的是:{logs:[节点A执行完成,节点B执行完成]}这时就需要Reducer。importoperatorfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictclassAgentState(TypedDict):logs:Annotated[list[str],operator.add]operator.add相当于告诉 LangGraph这个字段收到新值时不要覆盖。 把旧值和新值相加。Annotated[list[str], operator.add] 这个字段是字符串列表多个节点写入时用 拼接不覆盖。完整代码importoperatorfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassAgentState(TypedDict):logs:Annotated[list[str],operator.add]defnode_a(state:AgentState):return{logs:[A 执行完成]}defnode_b(state:AgentState):return{logs:[B 执行完成]}builderStateGraph(AgentState)builder.add_node(a,node_a)builder.add_node(b,node_b)builder.add_edge(START,a)builder.add_edge(a,b)builder.add_edge(b,END)graphbuilder.compile()resultgraph.invoke({logs:[]})print(result[logs])6. add_messages消息专用 Reducer聊天 Agent 一般会有{messages:[用户消息,AI消息,用户消息,工具消息,AI消息]}有时还需要根据 message ID 更新已有消息 删除消息 接受{role:user}这种简写 处理 HumanMessage、AIMessage、ToolMessage所以 LangGraph 提供了专用 Reduceradd_messagesfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlangchain.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messagesclassChatState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]7. MemorySaver thread_id多轮记忆首先记住add_messages 负责合并消息。Checkpointer 负责保存消息。thread_id 负责找到属于哪段对话的消息。缺一不可fromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlangchain.messagesimportAnyMessage,AIMessagefromlanggraph.graph.messageimportadd_messagesfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverclassChatState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]defchatbot(state:ChatState):message_countlen(state[messages])return{messages:[AIMessage(contentf我目前能看到{message_count}条历史消息。)]}builderStateGraph(ChatState)builder.add_node(chatbot,chatbot)builder.add_edge(START,chatbot)builder.add_edge(chatbot,END)# 创建内存 CheckpointermemoryMemorySaver()# 编译时传入 checkpointergraphbuilder.compile(checkpointermemory)# 同一段对话使用同一个 thread_idconfig{configurable:{thread_id:bells-chat-001}}# 第一轮result1graph.invoke({messages:[{role:user,content:你好我叫 Bells。}]},configconfig)print(result1[messages][-1].content)# 第二轮result2graph.invoke({messages:[{role:user,content:这是第二轮对话。}]},configconfig)print(result2[messages][-1].content)8. 长短期记忆8.1 短期记忆短期记忆依赖Checkpointer thread_id保存的是用户刚才说了什么工具调用了什么图执行到哪个节点当前任务进行到哪里8.2 长期记忆长期记忆是跨 thread 的。例如当前有两段不同对话thread_id rag-chat-001thread_id python-chat-003但两段对话都属于同一个用户user_id bells你希望 Agent 在两个不同对话中都知道用户主要学习 Python这时候需要Store user_id namespace9. interrupt让 Agent 停下来等人interrupt()的作用图先暂停 保存当前状态 把问题交给外部用户 等待用户输入 之后从暂停位置继续典型场景删除数据前确认 发送邮件前确认 调用付费接口前确认 执行 SQL 写操作前确认 发布文章前人工审核完整示例:fromtypingimportLiteralfromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportinterrupt,Commandfromlanggraph.checkpoint.memoryimportMemorySaverclassReviewState(TypedDict):draft:strstatus:NotRequired[str]defreview_node(state:ReviewState)-Command[Literal[publish,cancel]]:# 图会暂停在这里decisioninterrupt({question:是否发布这段内容,draft:state[draft]})# 恢复后decision 就是 Command(resume...) 传进来的值ifdecision[approved]:returnCommand(update{status:审核通过},gotopublish)returnCommand(update{status:审核拒绝},gotocancel)defpublish_node(state:ReviewState):print(发布内容,state[draft])return{status:已经发布}defcancel_node(state:ReviewState):return{status:已经取消}builderStateGraph(ReviewState)builder.add_node(review,review_node)builder.add_node(publish,publish_node)builder.add_node(cancel,cancel_node)builder.add_edge(START,review)builder.add_edge(publish,END)builder.add_edge(cancel,END)graphbuilder.compile(checkpointerMemorySaver())config{configurable:{thread_id:review-001}}# 第一次执行运行到 interrupt 后暂停pausedgraph.invoke({draft:这是待发布的内容。},configconfig)print(paused)# 用户审核通过后继续执行resultgraph.invoke(Command(resume{approved:True}),configconfig)print(result)