使用LangGraph的同时遇到了和Llamaindex相似的问题。这里是LangGraph的官方文档，可以多多参考其中的文档进行使用https://langchain-ai.github.io/langgraph/我使用的是本地vllm部署的LLM，所以没有配置API_KEY等信息。如果你使用的是API接口，可以参考官方文档中的方法来构建llm变量（还更方便）。比如这是一个例子：

def llm_client(content:str="你好，介绍一下你自己？"):
    url = "http://127.0.0.1:8000/v1/chat/completions"
    headers = {"Content-Type": "application/json", "Authorization": "Bearer "}
    data = {
        "model": "/workspace/models/Qwen3-0.6B",
        "messages": [{"role": "user", "content": content}],
        "temperature": 0.7,
    }

    response = requests.post(url, headers=headers, data=json.dumps(data))

    print("Status code:", response.status_code)

    try:
        result = response.json()
        # print("Response JSON:", result)
        return result
    except Exception as e:
        print("Failed to parse JSON:", e)
        return {}

在使用之前最好先pip这几个常用的包，可以避免很多之后的麻烦：

pip install langgraph
pip install langgraph-checkpoint-sqlite
pip install sqlite3


from langgraph.graph import StateGraph,START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, List,Optional
from datetime import datetime

基础应用篇

1. 在使用LangGraph之前首先需要定义节点之间传输的数据结构：

# 定义状态结构
class ChatState(TypedDict):
    messages: List[dict]
    metadata: Optional[dict]


 # 但是由于sqlite每次保存都会顺序保存其中的所有key，为了方便管理也可以统一写在一个tiao中
class ChatState(TypedDict):
    tiao:List[dict] # 在这里面写messages和metadata，或者使用一个BaseModel来定义tiao的结构

在这里我定义了一个ChatState的数据传输结构用于存储整个工作流中数据的传输。但是在使用sqlite数据库保存的时候会出现一个问题：该类中有多少个属性每一次保存数据库中就会分多少条，并且不会覆盖之前保存的信息，每一次都是增量保存。这就导致了我这里定义两个属性，在数据库里面每一次都会保存为两条数据（个人感觉还是很不好读取的），并且一轮对话就会从头到尾保存一次所有对话。

2. 定义LLM的回复方式，以及消息历史的保存方式：

在使用的过程中我遇到了很多问题，首先LangGraph并不会帮助你拼接对话历史，需要你自己决定需要保存什么。这个节点的作用就在于得到LLM的回复，并且自己定义处理对话历史的逻辑并保存到之前定义的State结构里。比如你可以这么做（如果出现问题可能是llm_client的输入和你的state["messages"]不匹配，可以打印出来修改一下即可）：

import re
def chatbot(state: ChatState):
    # print(state["messages"])
    # 先调用接口，拿到回复
    result = llm_client(messages=state["messages"]) # 如果出问题，可能是这里的参数格式没传对，修改一下即可
    reply = result["choices"][0]["message"]
    # 清洗content
    content = re.sub(r'<think>.*?</think>', '', reply.get('content', ''), flags=re.DOTALL).lstrip()
    return {
        "messages": state["messages"] + [{'role': reply.get('role'), 'content': content}],
        "metadata": {"timestamp": datetime.now().isoformat()}
    }

chatbot方法接收ChatState类作为输入，方法中调用llm_client获得回复以后，手动做好你想要的数据解析并且拼接起来，注意：return也必须要按照ChatState的数据结构进行返回。

3. 连接工作流的对话图，将各个模块整合在一起：

这里我使用的是本地的sqlite数据库（但是在之后的使用中存在一定的问题，比如之前定义State结构说的问题，将来会对这里的处理逻辑进行优化）。graph_builder将我们定义的模块连接起来，当然可以在其中添加更多的结构，不过统一的传递数据媒介是State，如果你想对数据做更多的处理，那么在之前的Sdate定义中就需要加上你需要的字段，并定义方法处理。

# 创建状态图
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
# 持久化到 SQLite
# 创建 SQLite 原生连接
import sqlite3
conn = sqlite3.connect("chat_history.db", check_same_thread=False)
saver = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=saver) # 这里指明了将数据持久化到哪里，不过我的建议是熟悉以后使用conn直接操作数据库，不要使用它的自动保存

可以看到在graph_builder中我使用add_edge把各个组件串联了起来，通过add_node添加节点（START和END时LangGraph自带的节点作为工作流的开始和结束）。需要注意的是，你可以通过add_node和之前定义chatbot类似的方法去添加更多的节点。将他们两两相连就可以组成一个专属工作流。最后通过.compile()来编译你的图结构，checkpointer参数用来指定持久化保存的路劲。

4. 定义工作流的输入，并且获得最终的回复：
定义一个处理用户提问的工作流，使用stream完成一次llm处理，assistant_msg中打印出最后结果（这个函数我没有使用return，你也可以根据实际需要选择需要return的内容）。

current_state = {"messages": [], "metadata": {}} # 在这里首先初始化状态
# 如果你已有数据库数据，可以使用你定义的方法导入
def stream_graph_updates(user_input: str):
    global current_state
    # 把用户消息加进对话历史
    current_state["messages"].append({"role": "user", "content": user_input})

    # 调用图，传入完整对话历史
    for event in graph.stream(current_state, config={"configurable": {"thread_id": "thread0"}}): 
    # 在这里面临着一个数据库自动保存的问题，LangGraph会在每一次调用stream的时候自动保存一次到数据库里，而且做的是添加操作而不是修改操作。这就导致如果用自动保存的话会出现非常多条记录，并且最后一条还保存着之前的记录，不太好用
        for value in event.values():
            assistant_msg = value["messages"][-1]
            print("Assistant:", assistant_msg)
            # 更新当前状态，保证上下文连贯
            current_state = value

可以看到其实最重要的是graph.stream()方法，当你调用一次这个方法后久相当于进行了一轮对话，graph会根据你的输入current_state走完一轮之前定义的流程图后得到最终的输入，并且原地更新current_state（所以这里我的current_state为全局变量）。

5. while True循环，拥有对话上下文记忆的LLM对话助手：
最后，运行下面这段代码，就可以获得一个拥有对话上下文的LLM助手啦。

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except Exception as e:
        print(f"An error:{e}. Please try again.")
        break

graph.invoke(current_state, config={"configurable": {"thread_id": "thread0"}})
# 使用这个可以将最后的结果保存到数据库中。

while True实时获取你的问题获得回复。最后的最后为了避免工作流错误导致时间丢失，可以使用invoke将全局变量current_state保存到数据库里，方便之后的读取。

6. 最后，从数据库中拿取已有的记录：

那么，在下一次加载时，因为保存的是二进制数据因此需要msgpack进行一次解码。你可以查看数据库中的数据保存的样式，来决定在你的程序中如何导入持久化的对话记录（下面的方法我没有使用metadata，你在使用的时候把前面的ChatState中的metadata去掉就可以了）。

import msgpack
def load_all_states_from_db(conn, thread_id="thread0"):
    cursor = conn.cursor()
    cursor.execute("""
        SELECT value FROM writes
        WHERE thread_id = ?
    """, (thread_id,))
    rows = cursor.fetchall()
    if not rows:
        print("没有找到任何记录")
        return []

    all_states = []
    for (binary_data,) in rows:
        if isinstance(binary_data, str):
            binary_data = binary_data.encode('latin1')
        try:
            unpacker = msgpack.Unpacker(raw=False)
            unpacker.feed(binary_data)
            for unpacked in unpacker:
                all_states.append(unpacked)
        except Exception as e:
            print("解包失败:", e)
            continue
    return all_states
all_states = load_all_states_from_db(conn, "thread0")
# print(f"总共加载到 {len(all_states)} 条记录")
# for i in range(1,6):
#     print(all_states[-i])

current_state = {"messages":all_states[-1]}

在这里我读取了最后一条数据，即之前保存的对话记录。在下一次启动程序的时候它依然能够记住你很久之前和它的聊天记录，从此，你拥有了一个具有长期记忆的LLM对话助手~