开源模型应用落地-LangGraph101-探索LangGraph短期记忆

一、前言

    在人工智能飞速发展的当下，智能体系统的设计不断迈向新高度。LangGraph 是一个强大的框架，专门用于构建和运行复杂的语言处理流程，其中，短期记忆管理宛如一座桥梁，连接着每一次对话的前后，对于对话质量和资源消耗起着举足轻重的作用。

    本文将深入探索 LangGraph 框架的短期记忆，以内存、sqlite以及mysql作为存储载体实现智能体的记忆功能。

 

---

二、术语介绍 2.1.LangGraph

     是一个基于 LangChain 构建的开源库，旨在为构建基于大语言模型（LLM）的有状态、多智能体应用程序提供一个全新的范式和强大的框架支持。

核心特点

支持循环图构建：与传统的基于有向无环图（DAG）的框架不同，LangGraph 允许开发者在图形结构中自由定义循环边和循环节点，这对于设计动态和迭代的智能体工作流程至关重要，能使智能体行为更贴近实际编程场景。自动状态管理：具有自动状态管理能力，可在多个交互中跟踪和持久化信息。随着智能体执行任务，状态会动态更新，确保系统能维护上下文并对新输入做出适当响应。多智能体协调：支持在单个图结构中协调多个智能体，每个智能体都可以有自己的提示、LLM、工具和自定义代码，方便构建多智能体系统，使多个智能体能够有效地协同工作，共同完成复杂任务。灵活性与可定制性：开发者可以灵活定义自己的智能体逻辑和通信协议，根据具体用例构建高度定制化的应用，满足不同场景需求。可扩展性与容错性：设计用于支持大规模多智能体应用的执行，具有强大的架构，能够处理大量交互和复杂工作流。同时，包含了优雅处理错误的机制，确保即使个别智能体遇到问题，应用也能继续运行。

工作原理

图结构表示：将应用视为一个有向图，其中每个节点代表一个 LLM 智能体或其他可执行的函数、工具等，边则定义了执行和数据流的方向，连接各个节点。状态更新机制：通过状态图来管理在执行周期中持久化的数据，当数据在节点之间流动时，状态对象会被管理和更新，从而实现跨执行周期的状态跟踪和更新。

核心组件

节点（Nodes）：是图结构中的执行单元，在其内部依据预设的代码逻辑进行复杂的运算与处理，最终输出一个经过更新后的全新状态。边（Edges）：边作为图结构中的连接要素，承担着构建节点之间逻辑关系与信息流向路径的重要职责。从类型上划分，主要涵盖了简单边和条件边这两种形式。简单边建立起节点之间最为基本的连接关系，使得前一个节点完成处理后所输出的状态信息能够顺畅地传输至下一个节点，成为其输入信息的来源，从而保障了信息在各个节点间的线性传递。而条件边则具备更为灵活和智能的特性，当一个节点的处理过程结束后，条件边会基于其内置的判断逻辑，精准地确定该节点后续应当通过哪些输出边来传递信息，极大地提高了系统的运行效率和任务处理的灵活性。状态（State）：作为一种共享的数据结构，其重要性不言而喻。在整个系统的运行过程中，状态如同一个公共的信息存储库，保存着应用程序在各个阶段的当前信息

应用场景

智能聊天机器人：能够处理各种用户请求，通过多 LLM 智能体协作，处理自然语言查询，提供准确响应，并在不同对话主题间无缝切换，提供连贯一致的用户体验。自主智能体：可创建根据用户输入和预定义逻辑执行任务的自主智能体，执行复杂工作流，与其他系统交互，并动态适应新信息，适用于自动化客户支持、数据处理、系统监控等任务。多智能体系统：在供应链管理、团队协作等场景中，多个智能体可分别管理库存、处理订单、协调交付等，通过 LangGraph 的协调能力，实现高效沟通、信息共享和决策同步，提升整体系统性能。工作流自动化工具：用于设计智能体来处理文档处理、审批工作流、数据分析等任务，通过定义清晰的工作流和利用状态管理，无需人工干预即可执行复杂操作序列，提高工作效率并减少错误。个性化推荐系统：利用多个智能体分析用户行为、偏好等数据，提供更个性化的推荐服务。

类似竞品

AutoGen：微软推出的行业先驱，核心设计包括用户智能体和助手智能体，特别适用于软件开发领域，在处理代码编排任务时表现出色。它允许用户在智能体间进行互动和手动调整，但对于没有编程背景的用户而言，学习成本较高。CrewAI：以简洁直观的操作界面受到推崇，非常适合快速搭建多智能体任务，用户无需深厚的技术背景，只需编写一些简短的提示词即可创建多个智能体并完成课程演示。不过，其在灵活性和定制化方面存在局限，技术社区的支持也相对较弱。OpenAI Swarm：专为新手设计，以简化智能体的创建与上下文切换为主要特征，用户只需简单操作即可快速制作演示应用。但它仅支持 OpenAI API，缺乏灵活性，不适合复杂的生产环境，技术社区支持也相对匮乏。Magentic-One：微软最新推出的产品，目标是为 AutoGen 提供一个简化的替代方案，尤其适合缺乏编程经验的用户，内置多个智能体用于文件管理、网页浏览和代码编写等功能，支持快速配置和使用。然而，其在对开源 LLMs 的支持上相对复杂，目前的文档和社区支持也未完善。

---

三、前置条件 3.1.基础环境及前置条件

  1. 操作系统：无限

3.2.安装依赖 conda create --name langgraph python=3.10 conda activate langgraph pip install langgraph -i pypi.tuna.tsinghua.edu /simple pip install langchain_openai langchain_community -i pypi.tuna.tsinghua.edu /simple pip install pillow -i pypi.tuna.tsinghua.edu /simple # sqlite示例需求 pip install langgraph-checkpoint-sqlite -i pypi.tuna.tsinghua.edu / # mysql示例需求 pip install langgraph-checkpoint-mysql pymysql cryptography -i pypi.tuna.tsinghua.edu / 3.3.安装mysql

下载地址： downloads.mysql /archives/community/

     1.下载完成后，解压文件

     2.解压后进行根目录，并创建data目录和my.ini文件

 

    my.ini文件内容如下，注意文件路径需要\\，否则会出现Can‘t find error-message file问题

[mysqld] # 设置3306端口 port=3306 # 设置mysql的安装目录 ---是你的文件路径--- basedir=E:\\software\\mysql8.0.40 # 设置mysql数据库的数据的存放目录 ---是你的文件路径data文件夹自行创建--- datadir=E:\\software\\mysql8.0.40\\data # 允许最大连接数 max_connections=200 # 允许连接失败的次数。 max_connect_errors=10 # 服务端使用的字符集默认为utf8mb4 character-set-server=utf8mb4 # 创建新表时将使用的默认存储引擎 default-storage-engine=INNODB # 默认使用“mysql_native_password”插件认证 #mysql_native_password default_authentication_plugin=mysql_native_password [mysql] # 设置mysql客户端默认字符集 default-character-set=utf8mb4 [client] # 设置mysql客户端连接服务端时默认使用的端口 port=3306 default-character-set=utf8mb4 [mysqld] # 用户密码的生命周期，设置为0 （不过期） default_password_lifetime=0

    3.使用管理员打开黑窗口（CMD），安装MySQL服务

    4.初始化MySQL生成随机密码（初始化mysql,这时候会生成一个密码一定要记清楚）

    5.启动MySQL服务

    6.首次登录MySQL

    输入mysql -u root -p点击回车，回车后输入刚刚牢记的密码

    7.修改root密码

ALTER USER 'root'@'localhost' IDENTIFIED BY '123456';

---

 

四、技术实现

4.1.内存示例 # -*- coding:utf-8 -*- import json import os import threading import traceback import requests from typing import Annotated, TypedDict from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition from pydantic import BaseModel, Field os.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxx'#你的Open AI Key gaode_api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxx' gaode_domain = " restapi.amap /v3" class ToolInput(BaseModel): region: str = Field(description="地区") @tool("get_current_weather", return_direct=True, args_schema=ToolInput) def get_current_weather(region:str): """根据传入的城市名查询天气""" ad_code = None try: session = requests.session() region_response = session.request( method="GET", url=f"{gaode_domain}/config/district?key={gaode_api_key}&keywords={region}&subdistrict=0", headers={"Content-Type": "application/json; charset=utf-8"}, ) region_response.raise_for_status() region_data = region_response.json() if region_data.get("info") == "OK": ad_code = region_data["districts"][0]["adcode"] except: traceback.print_exc() return f"获取{region}代码失败" try: if ad_code: weather_response = session.request( method="GET", url=f"{gaode_domain}/weather/weatherInfo?key={gaode_api_key}&city={ad_code}&extensions=all", headers={"Content-Type": "application/json; charset=utf-8"}, ) weather_response.raise_for_status() weather_data = weather_response.json() # print(f'weather_data: {weather_data}') if weather_data.get("info") == "OK": result = json.dumps(weather_data, ensure_ascii=False) # print(f'result: {result}') return json.dumps(weather_data,ensure_ascii=False) except: traceback.print_exc() return f"获取{region}天气预报信息失败" tools = [get_current_weather] # 定义状态类型 class State(TypedDict): messages: Annotated[list, add_messages] def chatbot(state: State): llm = ChatOpenAI(model="gpt-4o-mini",temperature=0,max_tokens=512) llm_with_tools = llm.bind_tools(tools) result = llm_with_tools.invoke(state["messages"]) return {"messages": [result]} def chat(input,thread_id): config = {"configurable": {"thread_id": thread_id}} for output in graph.stream({"messages": [("user", input)]}, config=config): for value in output.values(): print("Assistant:", value["messages"][-1].content) if __name__ == "__main__": # 创建并编译图 graph_builder = StateGraph(State) graph_builder.add_node("chatbot", chatbot) graph_builder.add_node("tools", ToolNode(tools=tools)) graph_builder.add_conditional_edges("chatbot", tools_condition) graph_builder.add_edge("tools", "chatbot") graph_builder.add_edge(START, "chatbot") memory = MemorySaver() graph = graph_builder pile(checkpointer=memory) thread_id = threading.get_ident() input = '今天广州的天气如何？' chat(input, thread_id) print('------------------------------------------------------------------') input = '刚才我问你哪里的天气情况了？' chat(input, thread_id)

调用结果：

本次执行的thread_id为：46196

修改上述代码，指定thread_id，看看智能体是否可以找回以前记忆？

 

4.2.sqlite示例

# -*- coding:utf-8 -*- import json import os import sqlite3 import threading import traceback import requests from typing import Annotated, TypedDict from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition from pydantic import BaseModel, Field os.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxx'#你的Open AI Key gaode_api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxx' gaode_domain = " restapi.amap /v3" class ToolInput(BaseModel): region: str = Field(description="地区") @tool("get_current_weather", return_direct=True, args_schema=ToolInput) def get_current_weather(region:str): """根据传入的城市名查询天气""" ad_code = None try: session = requests.session() region_response = session.request( method="GET", url=f"{gaode_domain}/config/district?key={gaode_api_key}&keywords={region}&subdistrict=0", headers={"Content-Type": "application/json; charset=utf-8"}, ) region_response.raise_for_status() region_data = region_response.json() if region_data.get("info") == "OK": ad_code = region_data["districts"][0]["adcode"] except: traceback.print_exc() return f"获取{region}代码失败" try: if ad_code: weather_response = session.request( method="GET", url=f"{gaode_domain}/weather/weatherInfo?key={gaode_api_key}&city={ad_code}&extensions=all", headers={"Content-Type": "application/json; charset=utf-8"}, ) weather_response.raise_for_status() weather_data = weather_response.json() # print(f'weather_data: {weather_data}') if weather_data.get("info") == "OK": result = json.dumps(weather_data, ensure_ascii=False) # print(f'result: {result}') return json.dumps(weather_data,ensure_ascii=False) except: traceback.print_exc() return f"获取{region}天气预报信息失败" tools = [get_current_weather] # 定义状态类型 class State(TypedDict): messages: Annotated[list, add_messages] def chatbot(state: State): llm = ChatOpenAI(model="gpt-4o-mini",temperature=0,max_tokens=512) llm_with_tools = llm.bind_tools(tools) result = llm_with_tools.invoke(state["messages"]) return {"messages": [result]} def chat(input,thread_id): config = {"configurable": {"thread_id": thread_id}} for output in graph.stream({"messages": [("user", input)]}, config=config): for value in output.values(): print("Assistant:", value["messages"][-1].content) if __name__ == "__main__": # 创建并编译图 graph_builder = StateGraph(State) graph_builder.add_node("chatbot", chatbot) graph_builder.add_node("tools", ToolNode(tools=tools)) graph_builder.add_conditional_edges("chatbot", tools_condition) graph_builder.add_edge("tools", "chatbot") graph_builder.add_edge(START, "chatbot") db_path = "./test.db" conn = sqlite3.connect(db_path, check_same_thread=False) memory = SqliteSaver(conn) graph = graph_builder pile(checkpointer=memory) thread_id = threading.get_ident() input = '今天广州的天气如何？' chat(input, thread_id) print('------------------------------------------------------------------') input = '刚才我问你哪里的天气情况了？' chat(input,thread_id)

调用结果：

本次执行的thread_id为：27552

生成的数据文件：

修改上述代码，指定thread_id，看看智能体是否可以找回以前记忆？

4.3.mysql示例 # -*- coding:utf-8 -*- import json import os import threading import traceback import pymysql import requests from typing import Annotated, TypedDict from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver from langgraph.graph import StateGraph, START from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition from pydantic import BaseModel, Field os.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxx'#你的Open AI Key gaode_api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxx' gaode_domain = " restapi.amap /v3" class ToolInput(BaseModel): region: str = Field(description="地区") @tool("get_current_weather", return_direct=True, args_schema=ToolInput) def get_current_weather(region:str): """根据传入的城市名查询天气""" ad_code = None try: session = requests.session() region_response = session.request( method="GET", url=f"{gaode_domain}/config/district?key={gaode_api_key}&keywords={region}&subdistrict=0", headers={"Content-Type": "application/json; charset=utf-8"}, ) region_response.raise_for_status() region_data = region_response.json() if region_data.get("info") == "OK": ad_code = region_data["districts"][0]["adcode"] except: traceback.print_exc() return f"获取{region}代码失败" try: if ad_code: weather_response = session.request( method="GET", url=f"{gaode_domain}/weather/weatherInfo?key={gaode_api_key}&city={ad_code}&extensions=all", headers={"Content-Type": "application/json; charset=utf-8"}, ) weather_response.raise_for_status() weather_data = weather_response.json() # print(f'weather_data: {weather_data}') if weather_data.get("info") == "OK": result = json.dumps(weather_data, ensure_ascii=False) # print(f'result: {result}') return json.dumps(weather_data,ensure_ascii=False) except: traceback.print_exc() return f"获取{region}天气预报信息失败" tools = [get_current_weather] # 定义状态类型 class State(TypedDict): messages: Annotated[list, add_messages] def chatbot(state: State): llm = ChatOpenAI(model="gpt-4o-mini",temperature=0,max_tokens=512) llm_with_tools = llm.bind_tools(tools) result = llm_with_tools.invoke(state["messages"]) return {"messages": [result]} def chat(input,thread_id): config = {"configurable": {"thread_id": thread_id}} for output in graph.stream({"messages": [("user", input)]}, config=config): for value in output.values(): print("Assistant:", value["messages"][-1].content) if __name__ == "__main__": # 创建并编译图 graph_builder = StateGraph(State) graph_builder.add_node("chatbot", chatbot) graph_builder.add_node("tools", ToolNode(tools=tools)) graph_builder.add_conditional_edges("chatbot", tools_condition) graph_builder.add_edge("tools", "chatbot") graph_builder.add_edge(START, "chatbot") conn = pymysql.connect( host="127.0.0.1", user="root", password="123456", database="agi", port=3306, ) memory = PyMySQLSaver(conn) #首次调用需要执行setup()执行数据结构初始化 #memory.setup() graph = graph_builder pile(checkpointer=memory) thread_id = threading.get_ident() input = '今天广州的天气如何？' chat(input, thread_id) print('------------------------------------------------------------------') input = '刚才我问你哪里的天气情况了？' chat(input,thread_id)

调用结果：

数据库生成的结果：

---

五、附带说明 5.1.问题一：RuntimeError: ‘cryptography‘ package is required for sha256_password

原因：缺少cryptography依赖包 解决方法：

pip install cryptography