langgraph_database_backend.py

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph.message import add_messages
from dotenv import load_dotenv
import sqlite3

load_dotenv()

llm = ChatOpenAI()

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def chat_node(state: ChatState):
    messages = state['messages']
    response = llm.invoke(messages)
    # print('response', response)
    return {"messages": [response.content]}


conn = sqlite3.connect(database='chatbot.db', check_same_thread=False)

# Checkpointer
checkpointer = SqliteSaver(conn=conn)

graph = StateGraph(ChatState)
graph.add_node("chat_node", chat_node)
graph.add_edge(START, "chat_node")
graph.add_edge("chat_node", END)

chatbot = graph.compile(checkpointer=checkpointer)


def retrieve_all_threads():
    all_threads = set()

    for checkpoint in checkpointer.list(None):
        all_threads.add(checkpoint.config['configurable']['thread_id'])

    return list(all_threads)

# print(chatbot)

# config = {
#     "configurable": {
#         "thread_id": "thread-test-1"
#     }
# }

# result = chatbot.invoke({ "messages": [HumanMessage(content="spell my name")]}, config=config)
# print('result', result)

# chatbot_stream = chatbot.stream(
#     {"messages": [HumanMessage(content="What is the recipe of paneer roll?")]},
#     config={"configurable": {"thread_id": "thread-test-streaming"}},
#     stream_mode="messages"
# )

# print('Streaming generator', chatbot_stream)

# for message_chunk, metadata in chatbot_stream:
#     print('message_chunk', message_chunk.content, end=' ', flush=True)