介绍
LangGraph 是langchain生态中低级别的编排框架，专用与构架流程化管理、长时间运行、有状态可持久化的智能系统。LangGraph给开发者提供对智能体更加细腻的控制。同时内置持久化执行、人工介入、记忆管理等特性。支持通过内存、SqlLite、PostgreSQL 等数据存储。程序重启可以读取已经存储的数据进行重新执行处理。
1.1 安装
通过pip即可安装，建议创建独立的虚拟环境管理依赖：
pip install langgraph langchain安装完成后，可以通过简单的导入测试验证环境是否正常：
fromlanggraph.graphimportStateGraph,STARTfromtyping_extensionsimportTypedDictclassState(TypedDict):message:str# 如果能正常导入并创建图对象，说明安装成功graph_builder=StateGraph(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 llm=init_chat_model(model="deepseek-v4-pro",model_provider="deepseek",api_key="sk-c62f66fa65xxxxxxx7be885b18bc",base_url="https://api.deepseek.com",extra_body={"thinking":{"type":"disabled"}},)classTestState(TypedDict):message:Annotated[Sequence[BaseMessage],add_messages]defnode_one(state:TestState):return{"message":[AIMessage(content="two")]}defnode_two(state:TestState):last_message=state["message"][-1]result=llm.invoke([last_message])return{"message":[result]}# 创建图并定义状态模式graph=StateGraph(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_compile=graph.compile()# 执行图result=graph.invoke({"message":[HumanMessage(content="你好")]})print(result["message"][-1].content)如果Langchain需要接入工具，可以如下这样操作：
# 定义工具@tooldefget_weather(city:str)->str:"""获取指定城市的天气"""returnf"{city}的天气总是晴朗的！"llm=init_chat_model(model="deepseek-v4-pro",model_provider="deepseek",api_key="sk-c62f66fa65xxxxxxx7be885b18bc",base_url="https://api.deepseek.com",extra_body={"thinking":{"type":"disabled"}},)# 绑定工具到模型llm_with_tools=llm.bind_tools([get_weather])defchatbot_node(state:State)->dict:response=llm_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(content="two")]}# 创建内存检查点保存器memory=InMemorySaver()# 创建工具节点tool_node=ToolNode(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)# 编译时传入检查点graph=graph_builder.compile(checkpointer=memory)# 执行带记忆的对话config={"configurable":{"thread_id":"conversation_1"}}# 第一次调用result1=graph.invoke({"messages":[HumanMessage(content="记住我的名字是 Alice")]},config=config)# 第二次调用，能记住上下文result2=graph.invoke({"messages":[HumanMessage(content="我叫什么名字？")]},config=config)执行图有多种输出模式：
同步阻塞：invoke
同步流式：stream
异步流式：astream
# 1. invoke：同步阻塞调用，返回最终结果result=graph.invoke(input,config)# 2. stream：流式输出中间状态foreventingraph.stream(input,config,stream_mode="values"):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_data=graph.get_graph().draw_mermaid_png()# 保存到文件withopen("graph.png","wb")asf:f.write(png_data)
