AIチャットボットの「記憶」を実現する階層メモリシステムの設計と実装
ももちゃんアプリに記憶を持たせるために軽量で多層的な記憶システムを実装しました。今回の実装もAIと共同作業でした。関数形式なので色々な用途に使えるはずです。ちなみにこの記事もAIが作成した技術ドキュメントをまとめたものです。
補足：9月7日
現在はこの記事の手法にさらに総合記憶を導入しています。短期、中期、長期の各記憶を入力として、600字程度で要約した記憶です。chatbotの回答の高速化を図るための入力Token低減化に寄与する手法です。重要な記憶のみが要約されているので通常の会話レベルでは問題なく過去の出来事を参照してくれます。
特徴
階層的メモリアーキテクチャ
普通によくある実装だと思います。長期間の会話の文脈を効率的に管理するための階層的メモリアーキテクチャです。人間の記憶システムににも似た構成で、短期記憶から長期記憶への段階的な情報集約を実現しています。最も短い記憶は、会話ログをLLMへのfewshotに組み入れることで実現しています。中期、長期記憶は心理学的にはエピソード記憶と言ったところです。
メモリ階層構造
┌─────────────────────────────────────────────────────────────────────┐
│ 会話メモリシステム階層図 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 【リアルタイム層】 │
│ ┌─────────────────┐ │
│ │ very_long_log │ ← 全ての会話ログ（最新50件を要約対象） │
│ │ (配列) │ │
│ └─────────────────┘ │
│ │ │
│ ▼ 50件ごとに要約 │
│ │
│ 【短期記憶層】（1日単位） │
│ ┌─────────────────┐ │
│ │ shortMemories │ ← 1日の会話を400文字で要約 │
│ │ (文字列) │ │
│ └─────────────────┘ │
│ │ │
│ ▼ 日付変更時に移動 │
│ │
│ 【中期記憶層】（1週間単位） │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ midMemories │ │ midLog │ │
│ │ (文字列) │ │ (配列: 最大7個) │ │
│ │ 1週間分の要約 │ │ 日次要約の蓄積 │ │
│ └─────────────────┘ └─────────────────┘ │
│ │ │ │
│ ▼ 要約統合 ▼ 7個を超えると移動 │
│ │
│ 【長期記憶層】（数週間〜数ヶ月） │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ longMemories │ │ longLog │ │
│ │ (文字列) │ │ (配列: 動的) │ │
│ │ 長期統合要約 │ │ 中期要約の蓄積 │ │
│ └─────────────────┘ └─────────────────┘ │
│ │ │
│ ▼ 定期的にクリア │
│ │
└─────────────────────────────────────────────────────────────────────┘リアルタイム層 (very_long_log)
データ型: 配列
容量: 無制限（実運用では定期的な整理が必要）
内容: 全ての会話ログ（ユーザー・アシスタント両方）
処理: 最新50件を抽出して短期記憶層に要約
📝 短期記憶層（shortMemories）
期間: 1日分の会話
容量: 約400文字
役割: その日の会話を要約して保持
📚 中期記憶層（midMemories + midLog）
期間: 1週間分（最大7日）
容量: 約400文字 + ログ7件
役割: 週単位での記憶を管理
🗄️ 長期記憶層（longMemories + longLog）
期間: 1-2年分（最大52件）
容量: 約600文字 + ログ52件
役割: 長期的なコンテキストを保持
シームレスな短期記憶
直近の50会話を要約して作成しています。この方法だと日時をまたいだ会話も途切れることなく要約して記憶できます。
記憶をセッションストアに統合
ももちゃんアプリはwebアプリなので、複数のクライアントからアクセス出来ます。そのためにクリアントごとにセッション情報を保持していますが、記憶もセッション情報の一部として保存されます。これによりクライアント毎に会話の記憶が異なり、ユーザーからは一貫性が保たれたように見えます。
データ構造
session_store内の構造
session_store[client_id] = {
"last_activity": "2025-08-13 15:30:00", # 最終活動時刻
# メモリストア（要約されたメモリ）
"memory_store": {
"shortMemories": "[2025-08-13 15:30:00] 今日はPythonの学習について...",
"midMemories": "[2025-08-12 10:00:00] 先週は機械学習の基礎を学習...",
"longMemories": "[2025-08-01 09:00:00] 7月はプログラミング全般について..."
},
# 会話ログ（生ログと中間ログ）
"long_conversation_logs": {
# リアルタイム層
"very_long_log": [
{
"role": "user",
"content": "ユーザーメッセージ=こんにちは",
"timestamp": "2025-08-13 15:25:00"
},
{
"role": "assistant",
"content": "こんにちは！何かお手伝いできることはありますか？",
"timestamp": "2025-08-13 15:25:01"
}
# ... 最新50件が要約対象
],
# 中期記憶層の生ログ
"midLog": [
"2025-08-12の要約: ...",
"2025-08-11の要約: ...",
# ... 最大7個
],
# 長期記憶層の生ログ
"longLog": [
"第1週の統合要約: ...",
"第2週の統合要約: ...",
# ... 要約後にクリア
]
}
}ログエントリの構造
very_long_logのエントリ
{
"role": "user" | "assistant",
"content": "メッセージ内容",
"timestamp": "YYYY-MM-DD HH:MM:SS"
}ユーザーメッセージの特殊処理
ユーザーメッセージは「ユーザーメッセージ=」プレフィックス付きで保存
要約時はプレフィックスを除去して処理
🔄 メモリの流れ（データフロー）
実際のデータがどのように流れるかを順を追って説明します
graph TD
A[新しい会話] --> B[very_long_logに追加]
B --> C[process_conversation_memory呼び出し]
C --> D[check_and_handle_date_change]
D --> E{日付変更？}
E -->|Yes| F[日付変更処理]
E -->|No| G[ログ抽出処理]
F --> H[shortMemories → midLog移動]
H --> I{midLog > 7個？}
I -->|Yes| J[midMemories → longLog移動]
I -->|No| K[midLog要約作成]
J --> L[longLog全体要約]
L --> M[longMemories更新]
M --> N[longLogの容量管理: 52件を超えた場合は古いログを削除]
N --> K
K --> G
G --> O[very_long_logから50件抽出]
O --> P[要約生成]
P --> Q[shortMemories更新]以下のようなstepになります。
Step 1: リアルタイム会話
ユーザー: "今日はPythonを勉強しました"
AI: "素晴らしいですね！どの分野を学習されましたか？"
↓
very_long_log に全て記録Step 2: 直近要約（50件ごと）
very_long_log（50件）
↓ AI要約処理
shortMemories: "[2025-08-13] 今日はPythonの基礎文法について学習..."Step 3: 日付変更時の記憶移行
shortMemories → midLog に移動
midLog が7件を超えると → longLog に移動
longLog が52件を超えると → 古いものから削除⚙️ 技術的な実装ポイント
1. 非同期処理による高速化
AI要約処理は時間がかかるため、非同期で実行：
async def process_conversation_memory(client_id, session_store, ...):
# 1. 日付変更チェック（高速）
await check_and_handle_date_change(...)
# 2. ログ抽出（高速）
recent_logs = extract_recent_very_long_logs(...)
# 3. AI要約（時間がかかる）
await generate_and_save_summary(...)2. ロック機構による排他制御
# データの整合性を保証
with session_store_lock:
session_store[client_id]["memory_store"]["shortMemories"] = summary3. エラーハンドリング
AI APIの失敗に備えたロバストな設計：
try:
response = await chat_req(a_client, prompt)
# 要約成功
except Exception as e:
print(f"要約生成エラー: {e}")
# エラーでもシステムは継続2. 日付変更時の処理詳細
# 疑似コード
async def handle_date_change(current_date, last_date):
if current_date != last_date:
# 1. 前日のshortMemoriesを保存
old_short_memories = session_store["shortMemories"]
# 2. midLogに追加
current_mid_memories = session_store["midMemories"]
session_store["midLog"].append(current_mid_memories + "\n" + old_short_memories)
# 3. midLogオーバーフロー処理
if len(session_store["midLog"]) > 7:
# midMemoriesをlongLogに移動
session_store["longLog"].append(current_mid_memories)
# midLogを7個に切り詰め
session_store["midLog"] = session_store["midLog"][-7:]
# 4. longLog要約処理
if len(session_store["longLog"]) > 0:
longlog_content = "\n".join(session_store["longLog"])
summary = await ai_summarize(longlog_content, 600)
session_store["longMemories"] = summary
# longLogは52件まで保持（要約後もクリアしない）
if len(session_store["longLog"]) > 52:
# 古いログから削除（最新52件を保持）
session_store["longLog"] = session_store["longLog"][-52:]
# 5. midLog要約処理
latest_midlog = session_store["midLog"][-1]
summary = await ai_summarize(latest_midlog, 400)
session_store["midMemories"] = summary3. AI要約の仕様
shortMemories要約（1日の要約）
prompt = """
以下のテキストを400文字程度に日記形式で要約した文章を作成すること。
このテキストは一日のユーザーとアシスタントの会話ログです。
作成する要約は会話の分析ではなく、会話ログにある出来事をまとめた要約です。
特に固有名詞や、行動などは重視してください。
絵文字は無視してください。
要約するテキスト={conversation_text}
"""midMemories要約（1日の統合要約）
prompt = """
以下のテキストを400文字程度に日記形式で要約した文章を作成すること。
このテキストは一日のユーザーとアシスタントの会話ログです。
作成する要約は会話の分析ではなく、会話ログにある出来事をまとめた要約です。
特に固有名詞や、行動などは重視してください。
絵文字は無視してください。
要約するテキスト={midlog_content}
"""longMemories要約（数週間の統合要約）
prompt = """
以下のテキストを600文字程度に要約した文章を作成すること。
このテキストは数週間分のユーザーとアシスタントの会話の要約集です。
作成する要約は時系列順に整理し、重要な出来事や継続的なテーマ、
固有名詞などを重視してください。
絵文字は無視してください。
要約するテキスト={longlog_content}
"""実装詳細
主要ファイル
1. memory.py
# 主要関数
- process_conversation_memory() # メイン処理関数
- check_and_handle_date_change() # 日付変更処理
- extract_recent_very_long_logs() # ログ抽出
- _generate_and_save_summary() # 要約生成
- _convert_logs_to_text() # ログ→テキスト変換2. app_writer.py
# 統合箇所
# 会話ログ追加後にメモリ処理を呼び出し
from memory import process_conversation_memory
await process_conversation_memory(
client_id=client_id,
session_store=session_store,
session_store_lock=session_store_lock,
chat_req=chat_req,
a_client=a_client,
current_date=None # 自動取得
)設定可能なパラメータ
# memory.py内の調整可能な値
MAX_MIDLOG_ENTRIES = 7 # midLogの最大保持数（週数）
MAX_EXTRACT_LOGS = 50 # very_long_logからの抽出数
SHORT_SUMMARY_LENGTH = 400 # shortMemories要約文字数
LONG_SUMMARY_LENGTH = 600 # longMemories要約文字数使用方法
1. 基本的な使用方法
システムの初期化
# app_writer.pyで会話ログを追加後、自動的にメモリ処理が実行される
session_store[client_id]["long_conversation_logs"]["very_long_log"].extend([
{"role": "user", "content": "ユーザーメッセージ=...", "timestamp": "..."},
{"role": "assistant", "content": "...", "timestamp": "..."}
])
# メモリ処理の実行
await process_conversation_memory(
client_id=client_id,
session_store=session_store,
session_store_lock=session_store_lock,
chat_req=chat_req,
a_client=a_client
)手動での日付変更処理
# テスト環境などで特定の日付を指定する場合
await process_conversation_memory(
client_id=client_id,
session_store=session_store,
session_store_lock=session_store_lock,
chat_req=chat_req,
a_client=a_client,
current_date="2025-08-14" # 明示的に指定
)📊 パフォーマンス特性
テストプログラムによる性能測定結果：
| 処理内容 | 実行時間 | ボトルネック |
|-------------------------|--------------|------------------|
| ログ抽出 | 0.01秒 | - |
| テキスト変換 | 0.05秒 | 文字列処理 |
| AI要約（短期） | 1-2秒 | API応答時間 | LLMサーバの性能によります
| AI要約（長期） | 2-3秒 | API応答時間 | LLMサーバの性能によります
| 日付変更処理全体 | 3-5秒 | AI要約×2回 |
🧪 テスト
AIエージェントによるテスト
エージェントにコードを解析させてテストプログラムを生成しました。
テストのやりやすさを考慮したコードの変更は私が提案して作成しています。
全11テストを実装：
✅ ログテキスト変換
✅ 最新ログ抽出
✅ 日付変更検出
✅ midLogオーバーフロー
✅ longLog容量管理（NEW!）
✅ AI要約生成
✅ 空データ処理
✅ エラーハンドリング
✅ 型変換処理
✅ データ構造チェック
✅ 統合処理フロー
テスト実行方法
# スタンドアロンテスト（依存関係最小）
python run_memory_tests.py
# pytestテスト（詳細レポート）
pytest test_memory.py -v🖥️ ダッシュボードによる監視
システムの状態を可視化するWebダッシュボードも開発しました。これは記憶がセッションストアの一部になっていて、アプリ稼働中にもセッション管理をしたいからです。この機能によってアプリを止めずにユーザーの削除や記憶の修正などが可能です。以下は実際のダッシュボード画面です。
セッションIDをクリックすると詳細を見ることが出来来ます。
会話履歴、メモリ情報、長期会話ログ、メモリストアはアコーディオンメニューになっていて、開くと参照や編集が出来ます。
コードの構成
コードはセッションストアへのアクセウ用のエンドポイントをアプリのバックエンドであるFastAPIに設置して、GUIであるHTMLからアクセスしています。
FastAPI + HTMLによる監視画面
@app.get("/api/memory-stats")
async def get_memory_stats():
"""全クライアントのメモリ統計を取得"""
for client_id, client_data in session_store.items():
# 統計情報を計算
yield MemoryStats(
client_id=client_id,
very_long_log_count=len(logs.get("very_long_log", [])),
midlog_count=len(logs.get("midLog", [])),
longlog_count=len(logs.get("longLog", []))
)監視可能な項目
🔍 リアルタイム監視:
very_long_log 件数増加率
AI要約の成功率
処理実行時間
エラー発生頻度
📈 定期監視:
クライアント数の推移
総メモリ使用量
データベースサイズ
古いデータの蓄積状況
🚀 実際の使用方法
基本的な実装例
from memory import process_conversation_memory, check_and_handle_date_change
from openai_chat import chat_req
from openai import AsyncOpenAI
# AIクライアント初期化
a_client = AsyncOpenAI(
base_url="http://localhost:11434/v1", # Ollama使用時
api_key="your-api-key"
)
# 記憶処理の実行
await process_conversation_memory(
client_id="user123",
session_store=session_store,
session_store_lock=lock,
chat_req=chat_req,
a_client=a_client
)ダッシュボードの起動
python dashboard_server.py
# → http://localhost:8000 でアクセス📋 まとめ
この階層メモリシステムにより、以下を実現しました：
�� 長期会話の効率的な記憶管理
🏗️ 3層階層による段階的な情報圧縮
⚡ 非同期処理による高レスポンス
🛡️ 包括的エラーハンドリング
📊 リアルタイム状況監視
🧪 全機能の品質保証
人間のような「記憶」を持つAIチャットボットの実現に向けて、確実な一歩を踏み出せたと考えています。
🔗 コードとリソース
GitHubリポジトリ: animede/chat-memory-system
技術仕様書: `MEMORY_SYSTEM_GUIDE.md`
クイックスタート: `MEMORY_QUICK_START.md`
変更履歴: `MEMORY_CHANGELOG.md`
参考 memory.py
import datetime
import time
from typing import List, Dict, Any, Optional
async def check_and_handle_date_change(
client_id: str,
session_store: Dict[str, Any],
session_store_lock,
chat_req,
a_client,
current_date: str
) -> None:
"""
日付変更をチェックして古いメモリの移動と要約処理を行う
Args:
client_id: クライアントID
session_store: セッションストア
session_store_lock: セッションストアのロック
chat_req: チャットリクエスト関数
a_client: AIクライアント
current_date: 現在の日付（YYYY-MM-DD形式）
"""
# 現在の日付と最後の活動日を比較して日付変更を検出
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
# session_storeから最後の活動時間を取得
if client_id in session_store:
last_activity = session_store[client_id].get("last_activity", "")
if last_activity:
last_date = last_activity[:10] # YYYY-MM-DD部分を取得
# 日付変更時の処理：古いshortMemoriesをmidLogに移動し、midLogを要約してmidMemoriesに保存
if current_date != last_date:
print(f"デバッグ: 日付変更を検出 - 旧日付: {last_date}, 新日付: {current_date}")
# 前日のshortMemoriesをmidLogに追加（最大7個=1週間分を保持）
old_short_memories = session_store[client_id]["memory_store"].get("shortMemories", "")
if old_short_memories.strip():
# 前日のshortMemoriesをmidLogに追加
with session_store_lock:
current_mid_memories = session_store[client_id]["memory_store"].get("midMemories", "")
session_store[client_id]["long_conversation_logs"]["midLog"].append(current_mid_memories + "\n" + old_short_memories)
# midLogが7を超えた場合の処理
if len(session_store[client_id]["long_conversation_logs"]["midLog"]) > 7: # 1週間分のログを保持
# midMemoriesをlongLogに移動
if current_mid_memories.strip():
# longLogが存在しない場合は初期化
if "longLog" not in session_store[client_id]["long_conversation_logs"]:
session_store[client_id]["long_conversation_logs"]["longLog"] = []
session_store[client_id]["long_conversation_logs"]["longLog"].append(current_mid_memories)
# midLogを7個に切り詰め
session_store[client_id]["long_conversation_logs"]["midLog"] = session_store[client_id]["long_conversation_logs"]["midLog"][-7:]
# longLogが存在し、内容がある場合は要約を作成してlongMemoriesに書き込み
if ("longLog" in session_store[client_id]["long_conversation_logs"] and
len(session_store[client_id]["long_conversation_logs"]["longLog"]) > 0):
longlog_entries = session_store[client_id]["long_conversation_logs"]["longLog"]
# 全てのlongLogエントリを結合
longlog_content = "\n".join(longlog_entries)
if longlog_content.strip():
user_msg = (
"以下のテキストを600文字程度に要約した文章を作成すること。このテキストは数週間分のユーザーとアシスタントの会話の要約集です。"
"作成する要約は時系列順に整理し、重要な出来事や継続的なテーマ、固有名詞などを重視してください。"
"絵文字は無視してください。要約するテキスト=" + longlog_content
)
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
try:
response_log = await chat_req(a_client, user_msg, "あなたは賢いAIです。userのリクエストに必ず日本語で答えること。リクエストの回答以外は答えないこと。")
response_log = f"[{now}] {response_log}"
print("longLogの要約をlongMemoriesに保存:", response_log)
with session_store_lock:
# longMemoriesに要約を保存
session_store[client_id]["memory_store"]["longMemories"] = response_log
# longLogが52個（1年分）を超えた場合、古いログを削除
if len(session_store[client_id]["long_conversation_logs"]["longLog"]) > 52:
# 最新の52個を保持して古いものを削除
session_store[client_id]["long_conversation_logs"]["longLog"] = session_store[client_id]["long_conversation_logs"]["longLog"][-52:]
print(f"デバッグ: longLogが52個を超えたため、古いログを削除しました。現在の件数: {len(session_store[client_id]['long_conversation_logs']['longLog'])}")
except Exception as e:
print(f"longLog要約作成中にエラー: {e}")
# midLogの最新エントリを要約してmidMemoriesに書き込み
midlog_text = session_store[client_id]["long_conversation_logs"]["midLog"]
if isinstance(midlog_text, list) and len(midlog_text) > 0:
# midLogがリストの場合、最新のエントリを取得
midlog_content = midlog_text[-1] if midlog_text else ""
elif isinstance(midlog_text, str):
# midLogが文字列の場合、そのまま使用
midlog_content = midlog_text
else:
midlog_content = ""
if midlog_content.strip():
user_msg = (
"以下のテキストを400文字程度に日記形式で要約した文章を作成すること。このテキストは一日のユーザーとアシスタントの会話ログです。"
"作成する要約は会話の分析ではなく、会話ログにある出来事をまとめた要約です。特に固有名詞や、行動などは重視してください。"
"絵文字は無視してください。要約するテキスト=" + midlog_content
)
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
try:
response_log = await chat_req(a_client, user_msg, "あなたは賢いAIです。userのリクエストに必ず日本語で答えること。リクエストの回答以外は答えないこと。")
response_log = f"[{now}] {response_log}"
print("midLogの要約をmidMemoriesに保存:", response_log)
with session_store_lock:
# 既存のmidMemoriesに書き込む
current_mid_memories = session_store[client_id]["memory_store"].get("midMemories", "")
if current_mid_memories.strip():
session_store[client_id]["memory_store"]["midMemories"] = current_mid_memories + "\n" + response_log
else:
session_store[client_id]["memory_store"]["midMemories"] = response_log
except Exception as e:
print(f"midLog要約作成中にエラー: {e}")
async def process_conversation_memory(
client_id: str,
session_store: Dict[str, Any],
session_store_lock,
chat_req,
a_client,
current_date: Optional[str] = None
) -> List[Dict[str, Any]]:
"""
会話ログが最大数を超えた場合の記憶処理を行う関数
Args:
client_id: クライアントID
session_store: セッションストア
session_store_lock: セッションストアのロック
chat_req: チャットリクエスト関数
a_client: AIクライアント
current_date: 現在の日付（YYYY-MM-DD形式）。Noneの場合は自動取得
Returns:
なし ->session_storeが更新されているため、返り値は不要
"""
# current_dateが指定されていない場合は現在の日付を自動取得
if current_date is None:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
current_date = timestamp[:10] # YYYY-MM-DD部分を取得
# 1. 日付変更チェック処理を最初に実行
await check_and_handle_date_change(
client_id=client_id,
session_store=session_store,
session_store_lock=session_store_lock,
chat_req=chat_req,
a_client=a_client,
current_date=current_date
)
# 2. very_long_logから新しいログを抽出してテキスト変換
print("====================================================================================================")
# session_storeのlong_conversation_logsから最新のログを取得（ロック使用）
with session_store_lock:
if client_id in session_store:
# very_long_logから新しいログを50件抜き出してテキスト変換
recent_logs_text = extract_recent_very_long_logs(client_id, session_store, max_logs=50)
print("====================================================================================================")
print("デバッグ: very_long_logから新しいログを50件抜き出してテキスト変換:", recent_logs_text)
print("====================================================================================================")
# 3. 要約生成処理（ロック外で実行 - AI処理は時間がかかるため）
# 抽出した会話ログを要約してshortMemoriesに保存
if recent_logs_text:
await _generate_and_save_summary(
recent_logs_text, client_id, session_store, session_store_lock, chat_req, a_client
)
return
def _convert_logs_to_text(logs: List[Dict[str, Any]]) -> str:
"""
ログをテキスト形式に変換する
Args:
logs: 会話ログのリスト
Returns:
テキスト形式に変換されたログ
"""
removed_text = ""
for log in logs:
role_name = "ユーザー" if log["role"] == "user" else "アシスタント"
# ユーザーの場合は"ユーザーメッセージ="以降を抜き出す、アシスタントの場合はそのまま
if log["role"] == "user" and "ユーザーメッセージ=" in log['content']:
content = log['content'].split("ユーザーメッセージ=")[1]
else:
content = log['content']
timestamp = log.get('timestamp', '')
removed_text += f"{timestamp} {role_name}: {content}\n"
return removed_text
def extract_recent_very_long_logs(
client_id: str,
session_store: Dict[str, Any],
max_logs: int = 50
) -> str:
"""
session_store[client_id]["long_conversation_logs"]["very_long_log"]から
新しいログを最大max_logs件抜き出し、テキストに変換する
Args:
client_id: クライアントID
session_store: セッションストア
max_logs: 抜き出す最大ログ数（デフォルトは50）
Returns:
テキスト形式に変換されたログ
"""
try:
# very_long_logが存在しない場合は空リストを初期化
if "very_long_log" not in session_store[client_id]["long_conversation_logs"]:
session_store[client_id]["long_conversation_logs"]["very_long_log"] = []
very_long_log = session_store[client_id]["long_conversation_logs"]["very_long_log"]
# 文字列の場合は空のリストに置換
if isinstance(very_long_log, str):
session_store[client_id]["long_conversation_logs"]["very_long_log"] = []
very_long_log = []
# 新しいログを最大max_logs件抜き出し
recent_logs = very_long_log[-max_logs:] if len(very_long_log) > max_logs else very_long_log
# ログをテキスト形式に変換
converted_text = _convert_logs_to_text(recent_logs)
print(f"デバッグ: very_long_logから{len(recent_logs)}件のログを抽出してテキストに変換しました")
return converted_text
except Exception as e:
print(f"very_long_logからのログ抽出中にエラー: {e}")
return ""
async def _generate_and_save_summary(
memory_data_for_summary: str,
client_id: str,
session_store: Dict[str, Any],
session_store_lock,
chat_req,
a_client
) -> None:
"""
要約を生成してセッションストアに保存する
Args:
memory_data_for_summary: 要約対象データ
client_id: クライアントID
session_store: セッションストア
session_store_lock: セッションストアのロック
chat_req: チャットリクエスト関数
a_client: AIクライアント
"""
try:
# メモリデータから改行文字を削除してテキストを整形
summarize_text = memory_data_for_summary.replace("\n", "").replace("\r", "") # 改行を削除
if summarize_text.strip(): # 空でない場合のみ要約処理を実行
user_msg = (
"以下のテキストを400文字程度に日記形式で要約した文章を作成すること。このテキストは一日のユーザーとアシスタントの会話ログです。"
"作成する要約は会話の分析ではなく、会話ログにある出来事をまとめた要約です。特に固有名詞や、行動などは重視してください。"
"絵文字は無視してください。要約するテキスト=" + summarize_text
)
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
response_log = await chat_req(a_client, user_msg, "あなたは賢いAIです。userのリクエストに必ず日本語で答えること。リクエストの回答以外は答えないこと。")
response_log = f"[{now}] {response_log}"
print("YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY response_log=", response_log)
# 要約結果をsession_storeに保存（再度ロック使用）
with session_store_lock:
if client_id in session_store:
session_store[client_id]["memory_store"]["shortMemories"] = response_log
except Exception as e:
print(f"要約生成中にエラーが発生しました: {e}") # エラーが発生してもログ保存は続行
コメント