🎉

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