手撸一个 deepseek 聊天历史对话、多轮对话(ollama + deepseek + langchain + flask)

发布于:2025-03-21 ⋅ 阅读:(30) ⋅ 点赞:(0)

前言

为了让聊天更加智能化,体验更好,打算搞个聊天历史对话,让 deepseek 可以结合上下文回答。

一、准备环境

  1. ollama,用于在本地运行、部署和管理大型语言模型(LLMs)。
  2. deepseek 模型,本文用的 deepseek-r1:14b。
  3. langchain,大语言模型应用程序的开发框架,主要 python 实现。

二、开发思路

  1. 使用 langchain 的 SQLChatMessageHistory 模块管理聊天历史记录,每段对话一个 session_id。
  2. 将聊天历史记录存储在 sqlite 数据库。
  3. 设置聊天上下文最大次数,即多轮对话次数,达到之后清空聊天记录。

三、代码解读

  1. 初始化 llm
# 设置Ollama的主机和端口
os.environ["OLLAMA_HOST"] = "127.0.0.1"
os.environ["OLLAMA_PORT"] = "11434"


def get_llm():
    # 定义提示模板
    template = """
    你是一个专业的助手,请根据用户的问题,生成详细且专业的回答。
    用户的问题:{input}
    """

    prompt = ChatPromptTemplate.from_template(template)

    # 初始化Ollama模型
    model = OllamaLLM(model="deepseek-r1:14b")

    # 创建链
    llm = prompt | model

    return llm
  1. 初始化聊天历史
# 初始化聊天历史
def get_session_history(session_id):
    return SQLChatMessageHistory(session_id, "sqlite:///chat_history.db")
  1. 基于聊天历史对话
def start_chat_history(input, session_id):
    # 判断是否新起对话
    delete_chat_history(session_id)

    # 初始化llm
    llm = get_llm()

    # 创建带历史记录的链
    runnable_with_history = RunnableWithMessageHistory(
        llm,
        get_session_history,
        input_messages_key="input"  # 明确指定输入键
    )

    # 提问参数
    translation_params = {
        "input": input
    }

    # 执行调用
    return runnable_with_history.stream(
        translation_params,
        config={"configurable": {"session_id": session_id}}
    )
  1. 达到最大对话记录清空历史

先看下 chat_history.db 的表结构

在这里插入图片描述
session_id 是个字符串类型,别的系统调用时可以把用户的 user_id 之类的作为对话标识,代码如下:

# 查询对话历史
def query_by_session_id(session_id):
    conn = sqlite3.connect('chat_history.db')
    curs = conn.cursor()
    sql = 'SELECT count(1) FROM message_store WHERE session_id = '+ session_id
    curs.execute(sql)
    results = curs.fetchall()
    curs.close()
    conn.close()
    return results


# 删除对话历史
def delete_by_session_id(session_id):
    conn = sqlite3.connect('chat_history.db')
    curs = conn.cursor()
    sql = 'DELETE FROM message_store WHERE session_id = '+ session_id
    curs.execute(sql)
    conn.commit()
    curs.close()
    conn.close()


# 达到最大对话记录清空历史
def delete_chat_history(session_id):
    history_count = query_by_session_id(session_id)
    count = history_count[0][0]
    # 一个对话两条记录,一个用户,一个AI
    if count >= MAX_CHAT * 2:
        delete_by_session_id(session_id)

五、完整代码

import os

import sqlite3
from langchain_ollama.llms import OllamaLLM
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

# 多轮对话次数
MAX_CHAT = 3

# 设置Ollama的主机和端口
os.environ["OLLAMA_HOST"] = "127.0.0.1"
os.environ["OLLAMA_PORT"] = "11434"


def get_llm():
    # 定义提示模板
    template = """
    你是一个专业的助手,请根据用户的问题,生成详细且专业的回答。
    用户的问题:{input}
    """

    prompt = ChatPromptTemplate.from_template(template)

    # 初始化Ollama模型
    model = OllamaLLM(model="deepseek-r1:14b")

    # 创建链
    llm = prompt | model

    return llm


# 查询对话历史
def query_by_session_id(session_id):
    conn = sqlite3.connect('chat_history.db')
    curs = conn.cursor()
    sql = 'SELECT count(1) FROM message_store WHERE session_id = '+ session_id
    curs.execute(sql)
    results = curs.fetchall()
    curs.close()
    conn.close()
    return results


# 删除对话历史
def delete_by_session_id(session_id):
    conn = sqlite3.connect('chat_history.db')
    curs = conn.cursor()
    sql = 'DELETE FROM message_store WHERE session_id = '+ session_id
    curs.execute(sql)
    conn.commit()
    curs.close()
    conn.close()


# 达到最大对话记录清空历史
def delete_chat_history(session_id):
    history_count = query_by_session_id(session_id)
    count = history_count[0][0]
    # 一个对话两条记录,一个用户,一个AI
    if count >= MAX_CHAT * 2:
        delete_by_session_id(session_id)


# 初始化聊天历史
def get_session_history(session_id):
    return SQLChatMessageHistory(session_id, "sqlite:///chat_history.db")


def start_chat_history(input, session_id):
    # 判断是否新起对话
    delete_chat_history(session_id)

    # 初始化llm
    llm = get_llm()

    # 创建带历史记录的链
    runnable_with_history = RunnableWithMessageHistory(
        llm,
        get_session_history,
        input_messages_key="input"  # 明确指定输入键
    )

    # 提问参数
    translation_params = {
        "input": input
    }

    # 执行调用
    return runnable_with_history.stream(
        translation_params,
        config={"configurable": {"session_id": session_id}}
    )

stream = start_chat_history("风景", "1")
for chunk in stream:
        print(chunk, end="", flush=True)

六、测试效果

代码设置的 3 次记录,到达这个记录后清空开启新对话。

  1. 第一轮对话,输入”深圳旅游“,返回内容:

在这里插入图片描述

  1. 第二轮对话,输入 ”美食“,返回内容:

在这里插入图片描述

这里可以看到已经关联了上次对话

  1. 第三轮对话,输入 ”特产“,返回内容:
    在这里插入图片描述
  2. 最后一轮,测试是否清空之前 3 次对话记录,开启新的对话,输入内容 ”自驾游“,返回内容:

在这里插入图片描述

可以看到,已经跟之前对话没有关联。

其他优化空间:

  • 每次对话,AI 回答的生成摘要存储,而不是全部存储,节省空间。
  • 通过 flask 或其他框架提供 API,给别的系统调用。
  • 也可以让调用的系统自行保存历史记录,在对话接口传入聊天历史内容。
  • 试试 langchain 其他模块实现聊天历史对话

大家可以自行修改~


网站公告

今日签到

点亮在社区的每一天
去签到