LangGraph笔记：一、LangGraph介绍及安装

介绍LangGraph 是langchain生态中低级别的编排框架专用与构架流程化管理、长时间运行、有状态可持久化的智能系统。LangGraph给开发者提供对智能体更加细腻的控制。同时内置持久化执行、人工介入、记忆管理等特性。支持通过内存、SqlLite、PostgreSQL 等数据存储。程序重启可以读取已经存储的数据进行重新执行处理。1.1 安装通过pip即可安装建议创建独立的虚拟环境管理依赖pip install langgraph langchain安装完成后可以通过简单的导入测试验证环境是否正常fromlanggraph.graphimportStateGraph,STARTfromtyping_extensionsimportTypedDictclassState(TypedDict):message:str# 如果能正常导入并创建图对象说明安装成功graph_builderStateGraph(State)langgraph 可以独立使用但通常会配合Langchain的模型包装器、工具定义和提示词模板进行使用。1.2 第一个StateGraph图StateGraph是LangGraph核心抽象主要用于工作流有方向的图整体包括三个部分State(每个图的状态)、Node(节点)、Edge(边)每个图都有一个状态(State)在节点间传递。fromlangchain.chat_modelsimportinit_chat_modelfromtypingimportAnnotated,Sequencefromlangchain_core.messagesimportBaseMessage,AIMessage,HumanMessagefromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportadd_messages,StateGraphfromtyping_extensionsimportTypedDict llminit_chat_model(modeldeepseek-v4-pro,model_providerdeepseek,api_keysk-c62f66fa65xxxxxxx7be885b18bc,base_urlhttps://api.deepseek.com,extra_body{thinking:{type:disabled}},)classTestState(TypedDict):message:Annotated[Sequence[BaseMessage],add_messages]defnode_one(state:TestState):return{message:[AIMessage(contenttwo)]}defnode_two(state:TestState):last_messagestate[message][-1]resultllm.invoke([last_message])return{message:[result]}# 创建图并定义状态模式graphStateGraph(TestState)# 添加节点graph.add_node(node_one,node_one)graph.add_node(node_two,node_two)# 定义边从 START 到 node_one从 node_one 到 node_twograph.add_edge(START,node_one)graph.add_edge(node_one,node_two)graph.add_edge(node_two,END)# 编译图grapg_compilegraph.compile()# 执行图resultgraph.invoke({message:[HumanMessage(content你好)]})print(result[message][-1].content)如果Langchain需要接入工具可以如下这样操作# 定义工具tooldefget_weather(city:str)-str:获取指定城市的天气returnf{city}的天气总是晴朗的llminit_chat_model(modeldeepseek-v4-pro,model_providerdeepseek,api_keysk-c62f66fa65xxxxxxx7be885b18bc,base_urlhttps://api.deepseek.com,extra_body{thinking:{type:disabled}},)# 绑定工具到模型llm_with_toolsllm.bind_tools([get_weather])defchatbot_node(state:State)-dict:responsellm_with_tools.invoke(state[messages])return{messages:[response]}LangGraph接入工具,并且增加记忆存储可以如下fromlanggraph.prebuiltimportToolNode,tools_conditionfromlanggraph.checkpoint.memoryimportInMemorySaver# 定义工具tooldefget_weather(city:str)-str:获取指定城市的天气returnf{city}的天气总是晴朗的defnode_one(state:TestState):return{message:[AIMessage(contenttwo)]}# 创建内存检查点保存器memoryInMemorySaver()# 创建工具节点tool_nodeToolNode(tools[get_weather])# 添加到图graph_builder.add_node(tools,tool_node)# 条件边根据 LLM 输出决定下一步graph_builder.add_conditional_edges(node_one,tools_condition,# 如果有工具调用则路由到 tools否则结束[tools,END])# 工具执行后返回 node_onegraph_builder.add_edge(tools,node_one)graph_builder.add_edge(node_one,END)# 编译时传入检查点graphgraph_builder.compile(checkpointermemory)# 执行带记忆的对话config{configurable:{thread_id:conversation_1}}# 第一次调用result1graph.invoke({messages:[HumanMessage(content记住我的名字是 Alice)]},configconfig)# 第二次调用能记住上下文result2graph.invoke({messages:[HumanMessage(content我叫什么名字)]},configconfig)执行图有多种输出模式同步阻塞invoke同步流式stream异步流式astream# 1. invoke同步阻塞调用返回最终结果resultgraph.invoke(input,config)# 2. stream流式输出中间状态foreventingraph.stream(input,config,stream_modevalues):print(event)# 每次状态更新都会输出# 3. astream异步流式asyncforeventingraph.astream(input,config):print(event)stream_mode参数控制输出内容“values”输出完整状态“updates”只输出被更新的字段“debug”包含详细的调试信息配置对象config除了可以配置thread_id,还可以配置其他运行参数便于其他参数传递config{configurable:{thread_id:conversation_1,user_id:user_123,model_name:claude-3-5-sonnet},tags:[production,chatbot],metadata:{session_type:support}}1.3、图结构可视化输出理解LangGraph的输出帮助开发者直观地理解决策流程。importiofromPILimportImage# 获取图片数据png_datagraph.get_graph().draw_mermaid_png()# 保存到文件withopen(graph.png,wb)asf:f.write(png_data)