langgraph学习笔记

langgraph 三个元素：状态、节点、边

• 状态：对话历史记录、上下文数据、内部变量
• 节点：输入一个状态，输出另一个状态
• 边：基于当前状态，决定下一步

状态

fromtyping_extensionsimportTypedDictfromtypingimportAnnotatedfromlanggraph.graphimportadd_messagesclassMyState(TypedDict):messages:Annotated[list,add_messages]

定义一个键值对，类型为列表，作用是add_messages（保存历史记录）

流程图

fromlanggraph.graphimportStateGraph graph=StateGraph(MyState)

节点

fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model='gpt-4o',api_key="sk",base_url="https://xiaoai.plus/v1")defchatbot(state:MyState):return{'messages':[llm.invoke(state['messages'])]}

graph.add_node('chatbot',chatbot)

add_node添加节点，（名称，函数）

边

fromlanggraph.constantsimportSTART,END# 表示： 流程从start----> chatbotgraph.add_edge(START,'chatbot')# 表示： 流程从chatbot----> ENDgraph.add_edge('chatbot',END)

构建图

graph=graph.compile()

执行

defloop_graph_invoke(user_input:str):"""循环调用这个流程图，让AI可以一直和用户对话"""result=graph.invoke({'messages':[('user',user_input)]})print('AI机器人: ',result['messages'][-1].content)whileTrue:try:user_input=input('用户: ')ifuser_input.lower()in['q','exit','quit']:print('对话结束，拜拜！')breakelse:loop_graph_invoke(user_input)exceptExceptionase:print(e)

工具节点

fromlanggraph.prebuiltimportToolNode tool_node=ToolNode(tools=tools)graph.add_node('tools',tool_node)

条件边

fromlanggraph.prebuiltimporttools_condition# 根据智能体自动决策是否需要调用工具，graph.add_conditional_edges('agent',tools_condition)

入口节点

# 设置入口节点graph.set_entry_point('agent')

langgraph 记忆能力

持久性检查点，checkpointer、thread_id

fromlanggraph.checkpoint.memoryimportMemorySaver memory_checkpointer=MemorySaver()graph=graph.compile(checkpointer=memory_checkpointer)

thread_id=input('请输入一个sessionId:')config={"configurable":{"thread_id":thread_id}}# 执行这个工作流whileTrue:try:user_input=input('用户: ')ifuser_input.lower()in['q','exit','quit']:print('对话结束，拜拜！')breakelse:loop_graph_invoke(graph,user_input,config)exceptExceptionase:print(e)

调用工具前中断

graph=graph.compile(checkpointer=memory_checkpointer,interrupt_before=['tools'])

interrupt_after=["tools"]

查看状态

now_state=graph.get_state(config)print(now_state)

更新状态

graph.update_state(config=config,values={'messages':new_message})