langchainを使って記憶領域を作成してみた
LangChainというものが巷で流行っています。
どのようなものなのかというものを実装してみました。
やったこと
Memory領域をlangchainを使って実装
Agents機能で、フィードバックを行いユーザの意図した内容にするように
Chainsで、どのような内容を取るのかを判定
%pip install langchain langchain-openai langchain-community
import os
os.environ["OPENAI_API_KEY"]="your-key"
import sqlite3
import json
import os
from typing import List, Dict, Any
from langchain_community.chat_message_histries import ChantMessageHistory
from langchain_core.prompts import ChatPromptTemplete, MessagePlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
db_file = "chat_histry.db"
def setup_database(db_file: str):
# ここに会話履歴とユーザー嗜好を入れたい
conn = None
try:
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
# 直近会話テーブル
cursor.execute('''
CREATE TABLE IF NOT EXISTS conversation_histry (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
turn_number INTEGER NOT NULL,
user_message TEXT,
ai_response TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
# ユーザー嗜好テーブル作成
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_preferences (
session_id TEXT PRIMARY KEY,
preferences TEXT -- ユーザー情報を文字列として保存(最近の嗜好性を反映)
)
''')
conn.commit()
def save_conversation_turn(session_id: str, turn_number: int, user_message: str, ai_response: str):
conn = None
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO conversation_history (session_id, turn_number, user_message, ai_response)
VALUES (?, ?, ?, ?)
''', (session_id, turn_number, user_message, ai_response))
conn.commit()
def get_conversation_history(session_id: str) -> ChatMessageHistory:
conn = None
history = ChatMessageHistory()
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
cursor.exesute('''
SELECT user_message, ai_response
FROM conversation_history
WHERE session_id = ?
ORDER BY turn_number
''', (session_id,))
for user_msg, ai_msg in cursor.fetchall():
history.add_user_message(user_msg)
history.add_ai_message(ai_msg)
return history
def save_user_preferences(session_id: str, preferences: Dict[str, Any]):
Discussion