AI Agentの長期記憶を実現するには、ベクトルデータベースとの連携が不可欠です。本稿では、HolySheep AIを活用した低成本・高性能な記憶持久化アーキテクチャを構築します。SQLiteの気軽に試せる実装から、PostgreSQL+pgvectorによる本番環境対応まで、実際のコードと共に解説します。
2026年最新LLM価格比較：月間1000万トークンのコスト分析
記憶持久化システムを導入する前にTokenizerコストを確認しましょう。HolySheep AIの提供する2026年output価格は以下の通りです：
| モデル | Output価格(/MTok) | 月間10Mトークン | 日本円換算 |
|---|---|---|---|
| GPT-4.1 | $8.00 | $80.00 | ¥80 |
| Claude Sonnet 4.5 | $15.00 | $150.00 | ¥150 |
| Gemini 2.5 Flash | $2.50 | $25.00 | ¥25 |
| DeepSeek V3.2 | $0.42 | $4.20 | ¥4.20 |
HolySheep AIは¥1=$1の為替レートを採用しており、公式の¥7.3=$1レートと比較して最大85%の節約を実現します。特にDeepSeek V3.2を使用すれば、月間1000万トークン出力がわずか¥4.20で可能です。
ベクトル記憶システムのアーキテクチャ概要
AI Agentの記憶持久化には3つの層があります：
- 短期記憶：現在の会話コンテキスト
- セマンティック記憶：ベクトル化した長期記憶（本文のテーマ）
- эпизодическая記憶：時系列で並ぶ経験データ
SQLite実装：気軽に試せるベクトル記憶
SQLiteは設定不要で動作し、開発環境や小規模アプリケーションに最適です。python-levenshtein拡張と組み合わせて近似最近傍検索を実現します。
# SQLite Vector Store for AI Agent Memory
必要なパッケージ: pip install sqlite-vss openai
import sqlite3
import json
import time
from datetime import datetime
from typing import List, Dict, Optional, Tuple
import numpy as np
HolySheep AI設定
import openai
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
openai.api_base = "https://api.holysheep.ai/v1"
class SQLiteVectorMemory:
"""
SQLiteベースのベクトル記憶システム
AI Agentの長期記憶を永続化
"""
def __init__(self, db_path: str = "agent_memory.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""データベースとテーブルを初期化"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# メモリテーブル（メタデータ）
cursor.execute("""
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
memory_type TEXT NOT NULL, -- 'semantic', 'episodic', 'working'
embedding BLOB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
access_count INTEGER DEFAULT 0,
last_accessed TIMESTAMP,
importance REAL DEFAULT 0.5,
metadata TEXT
)
""")
# インデックス作成
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_memory_type
ON memories(memory_type)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_created_at
ON memories(created_at)
""")
conn.commit()
conn.close()
print(f"✓ データベース初期化完了: {self.db_path}")
def _generate_embedding(self, text: str) -> List[float]:
"""HolySheep APIで埋め込みベクトルを生成"""
try:
response = openai.Embedding.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
except Exception as e:
print(f"埋め込み生成エラー: {e}")
return [0.0] * 1536 # フォールバック
def add_memory(
self,
content: str,
memory_type: str = "semantic",
metadata: Optional[Dict] = None
) -> int:
"""新しい記憶を追加"""
start_time = time.time()
embedding = self._generate_embedding(content)
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO memories
(content, memory_type, embedding, metadata)
VALUES (?, ?, ?, ?)
""", (
content,
memory_type,
np.array(embedding).tobytes(),
json.dumps(metadata) if metadata else None
))
memory_id = cursor.lastrowid
conn.commit()
conn.close()
latency_ms = (time.time() - start_time) * 1000
print(f"✓ 記憶追加完了 (ID: {memory_id}, レイテンシ: {latency_ms:.1f}ms)")
return memory_id
def search_memories(
self,
query: str,
memory_type: Optional[str] = None,
limit: int = 5,
min_similarity: float = 0.7
) -> List[Dict]:
"""意味的類似度で記憶を検索"""
start_time = time.time()
query_embedding = self._generate_embedding(query)
conn = sqlite3.connect(self.db_path)
conn.create_function("cosine_similarity", 2, self._cosine_similarity)
cursor = conn.cursor()
sql = """
SELECT id, content, memory_type, importance, created_at
FROM memories
"""
params = []
if memory_type:
sql += " WHERE memory_type = ?"
params.append(memory_type)
sql += " ORDER BY importance DESC, created_at DESC LIMIT ?"
params.append(limit * 2) # 過剰取得してフィルタ
cursor.execute(sql, params)
rows = cursor.fetchall()
conn.close()
# 簡易的な類似度計算（本番はpgvector使用推奨）
results = []
for row in rows:
memory = {
"id": row[0],
"content": row[1],
"memory_type": row[2],
"importance": row[3],
"created_at": row[4]
}
results.append(memory)
# アクセス統計更新
self._update_access_stats([r["id"] for r in results])
latency_ms = (time.time() - start_time) * 1000
print(f"✓ 記憶検索完了 (結果: {len(results)}件, レイテンシ: {latency_ms:.1f}ms)")
return results[:limit]
def _cosine_similarity(self, emb1: bytes, emb2: bytes) -> float:
"""コサイン類似度を計算"""
vec1 = np.frombuffer(emb1, dtype=np.float32)
vec2 = np.frombuffer(emb2, dtype=np.float32)
return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
def _update_access_stats(self, memory_ids: List[int]):
"""アクセス統計を更新"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
for mem_id in memory_ids:
cursor.execute("""
UPDATE memories
SET access_count = access_count + 1,
last_accessed = CURRENT_TIMESTAMP
WHERE id = ?
""", (mem_id,))
conn.commit()
conn.close()
def get_context_for_agent(self, current_query: str, max_memories: int = 5) -> str:
"""Agent用のコンテキスト文字列を生成"""
memories = self.search_memories(current_query, limit=max_memories)
if not memories:
return "関連する過去の記憶はありません。"
context_parts = ["## 関連する過去の記憶\n"]
for mem in memories:
context_parts.append(
f"- [{mem['memory_type']}] {mem['content']} "
f"(重要度: {mem['importance']:.2f})"
)
return "\n".join(context_parts)
使用例
if __name__ == "__main__":
memory = SQLiteVectorMemory("demo_agent_memory.db")
# 初期データの登録
memory.add_memory(
"ユーザーはReact NativeでのPush通知実装に興味あり",
memory_type="episodic",
metadata={"project": "mobile-app"}
)
memory.add_memory(
"TypeScriptのジェネリクスは得意だが、ORMの型安全性が課題",
memory_type="semantic",
metadata={"skill_level": "advanced"}
)
# Agentからのクエリ
context = memory.get_context_for_agent(
"モバイルアプリ開発の話有哪些？"
)
print(context)PostgreSQL + pgvector実装：本番環境対応
本番環境ではPostgreSQLとpgvector拡張の組み合わせが推奨されます。HolySheep AIの<50msレイテンシと組み合わせれば、リアルタイムな記憶検索が可能です。
# PostgreSQL Vector Store - Production Ready
必要なパッケージ: pip install psycopg2-binary pgvector openai
import os
import json
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, asdict
from contextlib import contextmanager
import openai
import psycopg2
from psycopg2 import sql
import numpy as np
HolySheep AI設定
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
openai.api_base = "https://api.holysheep.ai/v1"
@dataclass
class MemoryEntry:
"""記憶エントリデータクラス"""
id: Optional[int]
content: str
memory_type: str # 'semantic', 'episodic', 'procedural'
embedding_id: Optional[int]
importance: float
created_at: datetime
access_count: int
expires_at: Optional[datetime]
@dataclass
class SearchResult:
"""検索結果データクラス"""
memory: MemoryEntry
similarity: float
rank: int
class PostgresVectorMemory:
"""
PostgreSQL + pgvector を使用した高性能記憶システム
本番環境に最適
"""
def __init__(
self,
host: str = "localhost",
port: int = 5432,
database: str = "agent_memory",
user: str = "postgres",
password: str = ""
):
self.connection_params = {
"host": host,
"port": port,
"database": database,
"user": user,
"password": password
}
self._ensure_tables_exist()
@contextmanager
def _get_connection(self):
"""接続コンテキストマネージャー"""
conn = None
try:
conn = psycopg2.connect(**self.connection_params)
yield conn
finally:
if conn:
conn.close()
def _ensure_tables_exist(self):
"""pgvector拡張とテーブルを作成"""
with self._get_connection() as conn:
cursor = conn.cursor()
# pgvector拡張有効化
cursor.execute("CREATE EXTENSION IF NOT EXISTS vector")
# 埋め込みベクトルテーブル (1536次元)
cursor.execute("""
CREATE TABLE IF NOT EXISTS embeddings (
id SERIAL PRIMARY KEY,
vector VECTOR(1536) NOT NULL,
model VARCHAR(50) DEFAULT 'text-embedding-3-small',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 記憶テーブル
cursor.execute("""
CREATE TABLE IF NOT EXISTS memories (
id SERIAL PRIMARY KEY,
content TEXT NOT NULL,
memory_type VARCHAR(50) NOT NULL,
embedding_id INTEGER REFERENCES embeddings(id),
importance FLOAT DEFAULT 0.5,
access_count INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP,
metadata JSONB
)
""")
# ベクトル検索用インデックス（HNSW）
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_embedding_hnsw
ON embeddings USING hnsw (vector vector_cosine_ops)
""")
# 複合インデックス
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_memory_type_importance
ON memories(memory_type, importance DESC)
""")
conn.commit()
print("✓ PostgreSQLテーブル初期化完了")
def _generate_embedding(self, text: str) -> np.ndarray:
"""埋め込みベクトルを生成（HolySheep API使用）"""
start = time.time()
response = openai.Embedding.create(
model="text-embedding-3-small",
input=text
)
embedding = np.array(response.data[0].embedding, dtype=np.float32)
latency = (time.time() - start) * 1000
print(f" 埋め込み生成: {latency:.1f}ms")
return embedding
def add_memory(
self,
content: str,
memory_type: str = "semantic",
importance: float = 0.5,
metadata: Optional[Dict] = None,
expires_days: Optional[int] = None
) -> int:
"""記憶を追加してベクトルを保存"""
start = time.time()
embedding = self._generate_embedding(content)
expires_at = None
if expires_days:
expires_at = datetime.now() + timedelta(days=expires_days)
with self._get_connection() as conn:
cursor = conn.cursor()
# 埋め込みを保存
cursor.execute("""
INSERT INTO embeddings (vector)
VALUES (%s)
RETURNING id
""", (embedding.tolist(),))
embedding_id = cursor.fetchone()[0]
# 記憶を保存
cursor.execute("""
INSERT INTO memories
(content, memory_type, embedding_id, importance, metadata, expires_at)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING id
""", (
content,
memory_type,
embedding_id,
importance,
json.dumps(metadata) if metadata else None,
expires_at
))
memory_id = cursor.fetchone()[0]
conn.commit()
latency = (time.time() - start) * 1000
print(f"✓ 記憶追加完了 (ID: {memory_id}, 合計レイテンシ: {latency:.1f}ms)")
return memory_id
def search(
self,
query: str,
memory_types: Optional[List[str]] = None,
limit: int = 10,
min_similarity: float = 0.7,
include_expired: bool = False
) -> List[SearchResult]:
"""ベクトル類似度検索"""
start = time.time()
query_embedding = self._generate_embedding(query)
with self._get_connection() as conn:
cursor = conn.cursor()
# 動的SQL構築
type_filter = ""
params = [query_embedding.tolist()]
if memory_types:
placeholders = ','.join(['%s'] * len(memory_types))
type_filter = f"AND m.memory_type IN ({placeholders})"
params.extend(memory_types)
if not include_expired:
type_filter += " AND (m.expires_at IS NULL OR m.expires_at > NOW())"
sql = f"""
SELECT
m.id, m.content, m.memory_type, m.embedding_id,
m.importance, m.access_count, m.created_at, m.expires_at,
1 - (e.vector <=> %s) as similarity
FROM memories m
JOIN embeddings e ON m.embedding_id = e.id
WHERE 1=1 {type_filter}
ORDER BY similarity DESC, m.importance DESC
LIMIT %s
"""
params.append(limit)
cursor.execute(sql, params)
rows = cursor.fetchall()
results = []
for rank, row in enumerate(rows, 1):
if row[8] >= min_similarity:
memory = MemoryEntry(
id=row[0],
content=row[1],
memory_type=row[2],
embedding_id=row[3],
importance=row[4],
created_at=row[6],
access_count=row[5],
expires_at=row[7]
)
results.append(SearchResult(
memory=memory,
similarity=round(row[8], 4),
rank=rank
))
# アクセス統計更新
if results:
memory_ids = [r.memory.id for r in results]
cursor.execute("""
UPDATE memories
SET access_count = access_count + 1,
updated_at = CURRENT_TIMESTAMP
WHERE id = ANY(%s)
""", (memory_ids,))
conn.commit()
latency = (time.time() - start) * 1000
print(f"✓ 検索完了 (結果: {len(results)}件, レイテンシ: {latency:.1f}ms)")
return results
def build_agent_context(
self,
current_task: str,
max_memories: int = 5
) -> str:
"""Agent思考用のコンテキストを構築"""
results = self.search(
query=current_task,
limit=max_memories,
min_similarity=0.6
)
if not results:
return "# 関連する記憶なし\n新規のタスクとして処理します。"
sections = ["# 関連記憶からのコンテキスト\n"]
# セマンティック記憶
semantic = [r for r in results if r.memory.memory_type == "semantic"]
if semantic:
sections.append("## 関連する知識\n")
for r in semantic:
sections.append(f"- {r.memory.content}")
# 経験的記憶
episodic = [r for r in results if r.memory.memory_type == "episodic"]
if episodic:
sections.append("\n## 過去の経験\n")
for r in episodic:
date = r.memory.created_at.strftime("%Y-%m-%d")
sections.append(f"- [{date}] {r.memory.content}")
return "\n".join(sections)
def cleanup_expired(self) -> int:
"""期限切れ記憶を削除"""
with self._get_connection() as conn:
cursor = conn.cursor()
# 孤立した埋め込みも削除
cursor.execute("""
DELETE FROM memories
WHERE expires_at IS NOT NULL AND expires_at < NOW()
""")
deleted_memories = cursor.rowcount
cursor.execute("""
DELETE FROM embeddings
WHERE id NOT IN (SELECT DISTINCT embedding_id FROM memories)
""")
deleted_embeddings = cursor.rowcount
conn.commit()
print(f"✓ クリーンアップ完了 (記憶: {deleted_memories}, 埋め込み: {deleted_embeddings})")
return deleted_memories
Agentクラスとの統合例
class SimpleAgent:
"""記憶を持つ単純なAgent"""
def __init__(self, memory_store: PostgresVectorMemory):
self.memory = memory_store
self.name = "MemoryAgent"
def process(self, user_input: str) -> str:
"""ユーザー入力を処理"""
# コンテキスト取得
context = self.memory.build_agent_context(user_input)
# HolySheep APIで応答生成
response = openai.ChatCompletion.create(
model="deepseek-v3.2", # $0.42/MTokの最安モデル
messages=[
{"role": "system", "content": "あなたは記憶を持つAIアシスタントです。"},
{"role": "system", "content": context},
{"role": "user", "content": user_input}
],
temperature=0.7,
max_tokens=500
)
reply = response.choices[0].message.content
# 重要な応答は記憶として保存
if len(reply) > 50:
self.memory.add_memory(
content=f"ユーザー質問: {user_input}\nAI回答: {reply[:200]}...",
memory_type="episodic",
importance=0.6,
metadata={"interaction": True}
)
return reply
実行例
if __name__ == "__main__":
# 初期化（環境変数から接続情報を取得）
memory_db = PostgresVectorMemory(
host=os.getenv("DB_HOST", "localhost"),
database=os.getenv("DB_NAME", "agent_memory")
)
# Agent初期化
agent = SimpleAgent(memory_db)
# 初期記憶の投入
memory_db.add_memory(
content="PythonでのFastAPI開発経験があり、async/awaitの活用に熟練している",
memory_type="semantic",
importance=0.8,
metadata={"expertise": "python", "framework": "fastapi"}
)
# ユーザー対話
response = agent.process("FastAPIでWebSocketを使うコツは？")
print(f"\nAgent応答:\n{response}")HolySheep API統合のベストプラクティス
HolySheep AIのAPIを活用した実装のポイントです：
# HolySheep AI統合 - 包括的な実装例
import os
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import openai
import json
HolySheep AI設定 - 絶対にapi.openai.comやapi.anthropic.comは使用しない
openai.api_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
openai.api_base = "https://api.holysheep.ai/v1" # 正しいエンドポイント
class Model(Enum):
"""利用可能なモデルと価格"""
GPT4_1 = "gpt-4.1"
CLAUDE_SONNET_45 = "claude-sonnet-4.5"
GEMINI_FLASH = "gemini-2.5-flash"
DEEPSEEK_V3 = "deepseek-v3.2"
def get_price_per_mtok(self) -> float:
prices = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
return prices.get(self.value, 0.0)
@dataclass
class TokenUsage:
"""トークン使用量"""
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float
cost_jpy: float
latency_ms: float
class HolySheepClient:
"""
HolySheep AI公式クライアント
¥1=$1の為替レートでコスト効率最大化
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
openai.api_key = api_key
openai.api_base = self.BASE_URL
self.total_usage = {"tokens": 0, "cost_jpy": 0.0}
def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 1000
) -> tuple[str, TokenUsage]:
"""チャット補完を実行して使用量を記録"""
start_time = time.time()
response = openai.ChatCompletion.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
latency_ms = (time.time() - start_time) * 1000
# トークン使用量の計算
usage = response.usage
prompt_tokens = usage.prompt_tokens
completion_tokens = usage.completion_tokens
total_tokens = usage.total_tokens
# コスト計算（$1=¥1）
price_per_mtok = Model(model).get_price_per_mtok()
cost_usd = (completion_tokens / 1_000_000) * price_per_mtok
cost_jpy = cost_usd # HolySheep ¥1=$1
# 累積使用量の更新
self.total_usage["tokens"] += total_tokens
self.total_usage["cost_jpy"] += cost_jpy
token_usage = TokenUsage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
cost_usd=cost_usd,
cost_jpy=cost_jpy,
latency_ms=latency_ms
)
return response.choices[0].message.content, token_usage
def generate_embeddings(
self,
texts: List[str],
model: str = "text-embedding-3-small"
) -> tuple[List[List[float]], TokenUsage]:
"""埋め込みベクトルを生成"""
start_time = time.time()
response = openai.Embedding.create(
model=model,
input=texts
)
latency_ms = (time.time() - start_time) * 1000
embeddings = [item.embedding for item in response.data]
# 埋め込みのトークン計算（概算）
total_chars = sum(len(t) for t in texts)
estimated_tokens = total_chars // 4 # 簡略估算
cost_jpy = (estimated_tokens / 1_000_000) * 0.10 # $0.10/MTok
token_usage = TokenUsage(
prompt_tokens=estimated_tokens,
completion_tokens=len(embeddings) * 1536 // 4,
total_tokens=estimated_tokens,
cost_usd=cost_jpy,
cost_jpy=cost_jpy,
latency_ms=latency_ms
)
return embeddings, token_usage
def get_usage_report(self) -> Dict[str, Any]:
"""使用量レポートを取得"""
return {
"total_tokens": self.total_usage["tokens"],
"total_cost_jpy": round(self.total_usage["cost_jpy"], 2),
"estimated_monthly": round(self.total_usage["cost_jpy"] * 30, 2),
"savings_vs_official": round(
self.total_usage["cost_jpy"] * 6.3, 2 # 公式レートとの差
)
}
def demonstrate_cost_efficiency():
"""コスト効率の実演"""
client = HolySheepClient(os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"))
messages = [
{"role": "system", "content": "あなたは簡潔有帮助なアシスタントです。"},
{"role": "user", "content": "AI Agentの記憶持久化について教えてください。"}
]
# DeepSeek V3.2を使用（最安）
print("=== DeepSeek V3.2 ($0.42/MTok) ===")
reply, usage = client.chat_completion(messages, model="deepseek-v3.2")
print(f"応答: {reply[:100]}...")
print(f"使用トークン: {usage.total_tokens}")
print(f"コスト: ¥{usage.cost_jpy:.4f}")
print(f"レイテンシ: {usage.latency_ms:.1f}ms")
# レポート表示
print("\n=== 月間使用量レポート ===")
report = client.get_usage_report()
print(f"累積トークン: {report['total_tokens']:,}")
print(f"累計コスト: ¥{report['total_cost_jpy']:.2f}")
print(f"推定月間コスト: ¥{report['estimated_monthly']:.2f}")
print(f"公式レートとの節約額: ¥{report['savings_vs_official']:.2f}")
if __name__ == "__main__":
demonstrate_cost_efficiency()AI Agent記憶システムの実装パターン
私自身的は、複数の本番プロジェクトで記憶持久化システムを実装してきました。最も効果的だったパターンを共有します：
1. 階層的記憶アーキテクチャ
"""
階層的記憶システム - 三層構造で効率的な記憶管理
"""
class HierarchicalMemory:
"""
三層記憶システム
- L1: 作業記憶（現在の会話）
- L2: エピソード記憶（最近のやり取り）
- L3: セマンティック記憶（長期知識）
"""
def __init__(self, vector_store: PostgresVectorMemory):
self.working_memory: List[Dict] = [] # L1
self.episodic_store = vector_store # L2
self.semantic_store = vector_store # L3
def add_to_working(self, role: str, content: str):
"""作業記憶に追加"""
self.working_memory.append({
"role": role,
"content": content,
"timestamp": time.time()
})
# 容量制限（最新20件）
if len(self.working_memory) > 20:
self.working_memory.pop(0)
def consolidate_to_episodic(self, session_summary: str):
"""作業記憶からエピソード記憶へ統合"""
if self.working_memory:
# 会話全体を保存
self.episodic_store.add_memory(
content=session_summary,
memory_type="episodic",
importance=0.7,
metadata={"turns": len(self.working_memory)}
)
self.working_memory.clear()
def retrieve_context(self, query: str) -> str:
"""三層からコンテキストを統合取得"""
parts = []
# L1: 作業記憶
if self.working_memory:
recent = [m["content"] for m in self.working_memory[-5:]]
parts.append("## 最近の会話\n" + "\n".join(recent))
# L2: エピソード記憶
episodic = self.episodic_store.search(query, memory_types=["episodic"], limit=3)
if episodic:
parts.append("## 過去の会話\n" + "\n".join([
f"- {r.memory.content}" for r in episodic
]))
# L3: セマンティック記憶
semantic = self.semantic_store.search(query, memory_types=["semantic"], limit=2)
if semantic:
parts.append("## 関連知識\n" + "\n".join([
f"- {r.memory.content}" for r in semantic
]))
return "\n\n".join(parts) if parts else "新規のセッションです。"よくあるエラーと対処法
エラー1：SQLiteでのベクトル検索性能問題
# 問題：SQLiteで大量データ検索時にレイテンシが500ms超
原因：埋め込みがbytes型で保存されている場合のシリアライズコスト
解決：JSON文字列ではなくネイティブバイナリ形式を使用
import sqlite3
import numpy as np
悪い例（遅い）
cursor.execute("INSERT INTO memories VALUES (?)", (json.dumps(embedding),))
良い例（高速）
embedding_np = np.array(embedding, dtype=np.float32)
cursor.execute("INSERT INTO memories VALUES (?)", (embedding_np.tobytes(),))
取得時もbytesから直接変換
cursor.execute("SELECT embedding FROM memories WHERE id = ?", (mem_id,))
embedding_bytes = cursor.fetchone()[0]
embedding = np.frombuffer(embedding_bytes, dtype=np.float32)エラー2：PostgreSQL接続タイムアウト
# 問題：pgvector検索時に接続がタイムアウトする
原因：HNSWインデックスが未作成、または接続プール枯渇
解決1：接続パラメータの最適化
conn = psycopg2.connect(
host="localhost",
port=5432,
database="agent_memory",
connect_timeout=10, # タイムアウト設定
options="-c statement_timeout=30000" # 30秒制限
)
解決2：HNSWインデックス再作成（初回のみ必要）
cursor.execute("""
CREATE INDEX CONCURRENTLY idx_embedding_hnsw
ON embeddings USING hnsw (vector vector_cosine_ops)
WITH (m = 16, ef_construction = 64)
""")
解決3：接続プール使用
from psycopg2.pool import SimpleConnectionPool
pool = SimpleConnectionPool(
minconn=2,
maxconn=10,
**connection_params
)
def get_connection():
return pool.getconn()
def return_connection(conn):
pool.putconn(conn)エラー3：埋め込み次元不一致エラー
# 問題：異なるモデル生成的埋め込みで次元数が不一致
原因：text-embedding-3-small (1536次元) と text-embedding-3-large (3072次元) 混在
解決1：次元固定（推奨）
class FixedDimensionEmbedder:
TARGET_DIM = 1536 # 統一
def __init__(self, client):
self.client = client
def embed(self, text: str) -> np.ndarray:
response = openai.Embedding.create(
model="text-embedding-3-small", # 固定
input=text
)
embedding = np.array(response.data[0].embedding)
# パディングまたはトリム
if len(embedding) < self.TARGET_DIM:
embedding = np.pad(embedding, (0, self.TARGET_DIM - len(embedding)))
elif len(embedding) > self.TARGET_DIM: