AIエージェントが「以前何があったか忘れる」問題に頭を悩ませていませんか？私はHolySheep AIを使って複数のプロジェクトでエージェント開発していますが、コンテキストウィンドウの制限と長期記憶の両立は永遠のテーマでした。本稿では、ベクター検索とナレッジグラフを組み合わせたハイブリッド長期記憶システムの構築方法を、API経験が全くない初心者でも理解できるようにゼロから解説します。
なぜハイブリッド記憶システムが必要なのか
AIエージェントの記憶には3つの課題があります。ベクター検索だけでは「関連性は分かるが関係性の深さが分からない」、ナレッジグラフだけでは「構造は分かるが曖昧な検索に弱い」。私は以前、ベクター検索だけで顧客サポートエージェントを作ったところ、「類似の問い合わせ」を返すものの因果関係を理解できず、同じ質問を違う表現でされると途端に正確な回答ができなくなりました。
3つの記憶タイプの比較
+------------------+-------------------+-------------------+------------------+
| 記憶タイプ | 検索速度 | 関係性の表現 | 実装難易度 |
+------------------+-------------------+-------------------+------------------+
| 純粋ベクターDB | ~10-50ms | △ (コサイン類似度) | ○ (易于実装) |
| 純粋ナレッジグラフ| ~5-20ms | ◎ (三重項構造) | △ (複雑な設計) |
| ハイブリッド | ~30-80ms | ◎ (両方の強み) | ○ (段階的実装可) |
+------------------+-------------------+-------------------+------------------+HolySheep AIのAPIは<50msのレイテンシを提供しているため、ハイブリッド検索のオーバーヘッドを気にせず実装できます。¥1=$1というレート（公式¥7.3=$1比85%節約）で、大量のリクエストを送るプロジェクトでもコストを押さえられます。
アーキテクチャ設計
ハイブリッド記憶システムのアーキテクチャは以下の3層で構成されます：
+------------------------------------------------------------------+
| アプリケーション層 |
| ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ |
| │ エージェント │ │ 検索API │ │ 記憶更新サービス │ |
| └─────────────┘ └─────────────┘ └─────────────────────────┘ |
+------------------------------------------------------------------+
| 記憶統合層 |
| ┌─────────────────────┐ ┌──────────────────────────────┐ |
| │ ベクターDB検索 │ │ ナレッジグラフクエリ │ |
│ │ (セマンティック) │ │ (関係性・経路探索) │ |
| └─────────────────────┘ └──────────────────────────────┘ |
| │ 融合スコアリング |
+------------------------------------------------------------------+
| ストレージ層 |
| ┌─────────────────────┐ ┌──────────────────────────────┐ |
| │ PostgreSQL+pgvector │ │ Neo4j / Memgraph │ |
| │ (埋め込みベクトル) │ │ (ノード・エッジ) │ |
| └─────────────────────┘ └──────────────────────────────┘ |
+------------------------------------------------------------------+実装：Step-by-Step
Step 1：プロジェクト準備と環境構築
まず、必要なライブラリをインストールします。Python環境がない場合は、Python 3.9以上をインストールしてください。
mkdir agent-memory-system
cd agent-memory-system
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
必要なライブラリをインストール
pip install openai psycopg2-binary neo4j python-dotenv requests💡 ヒント：エラーが表示された場合、pip install --upgrade pipを実行してpipを最新版に更新してください。
Step 2：環境変数の設定
# .envファイルを作成
cat > .env << 'EOF'
HolySheep AI設定
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
データベース設定
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=agent_memory
POSTGRES_USER=your_user
POSTGRES_PASSWORD=your_password
Neo4j設定
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=your_neo4j_password
EOF💡 スクリーンショットイメージ：.envファイルの内容がテキストエディタで開かれており、APIキーとデータベース接続情報が設定済み。
Step 3：埋め込み生成クライアントの実装
HolySheep AIを使ってテキストからベクトル埋め込みを生成します。私は実際にこの方法で顧客の会話を埋め込んでいますが、DeepSeek V3.2の埋め込みコスト($0.42/MTok)が特に経済的です。
import os
import requests
from dotenv import load_dotenv
load_dotenv()
class HolySheepEmbedder:
"""HolySheep AI用于生成テキスト埋め込み"""
def __init__(self):
self.api_key = os.getenv("HOLYSHEEP_API_KEY")
self.base_url = os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
self.model = "text-embedding-3-small" # 高精度・低コスト
def get_embedding(self, text: str) -> list[float]:
"""単一テキストの埋め込みベクトルを取得"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input": text,
"model": self.model
}
)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
def get_embeddings_batch(self, texts: list[str]) -> list[list[float]]:
"""複数テキストの埋め込みを一括取得"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input": texts,
"model": self.model
}
)
response.raise_for_status()
return [item["embedding"] for item in response.json()["data"]]
テスト実行
if __name__ == "__main__":
embedder = HolySheepEmbedder()
sample_text = "顧客が配送遅延について苦情を言った"
embedding = embedder.get_embedding(sample_text)
print(f"埋め込み次元数: {len(embedding)}")
print(f"最初の5次元: {embedding[:5]}")
# コスト確認（実際の価格は$0.42/MTok）
print(f"推定コスト: ${len(sample_text) / 1_000_000 * 0.42:.6f}")💡 スクリーンショットイメージ：コンソールに「埋め込み次元数: 1536」「最初の5次元: [0.0231, -0.0894, 0.0567, ...]」と表示。
Step 4：ナレッジグラフクライアントの実装
Neo4jを使ってエンティティ間の関係性を表現します。顧客名、製品、アクション、イベントをノードとして接続し因果関係を描きます。
from neo4j import GraphDatabase
class KnowledgeGraphClient:
"""ナレッジグラフ操作用クライアント"""
def __init__(self, uri: str, user: str, password: str):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def add_memory_node(self, entity_type: str, properties: dict):
"""記憶ノードを追加"""
with self.driver.session() as session:
session.execute_write(self._create_node, entity_type, properties)
@staticmethod
def _create_node(tx, entity_type: str, properties: dict):
query = f"""
CREATE (n:{entity_type} {{id: $id, created_at: datetime()}})
SET n += $properties
RETURN n
"""
tx.run(query, id=properties.get("id", ""), properties=properties)
def add_relationship(self, from_id: str, to_id: str,
relationship_type: str, properties: dict = None):
"""ノード間の関係を追加"""
with self.driver.session() as session:
session.execute_write(
self._create_relationship, from_id, to_id,
relationship_type, properties or {}
)
@staticmethod
def _create_relationship(tx, from_id: str, to_id: str,
rel_type: str, properties: dict):
query = f"""
MATCH (a {{id: $from_id}}), (b {{id: $to_id}})
CREATE (a)-[r:{rel_type} $properties]->(b)
RETURN r
"""
tx.run(query, from_id=from_id, to_id=to_id,
properties=properties)
def find_related_entities(self, entity_id: str, depth: int = 2) -> list[dict]:
"""指定ノードに関連するエンティティを取得（経路探索）"""
with self.driver.session() as session:
result = session.execute_read(
self._find_path_entities, entity_id, depth
)
return result
@staticmethod
def _find_path_entities(tx, entity_id: str, depth: int) -> list[dict]:
query = f"""
MATCH path = (start {{id: $entity_id}})-[*1..{depth}]-(connected)
RETURN path,
length(path) as distance,
type(last(relationships(path))) as last_relation
ORDER BY distance
LIMIT 20
"""
result = tx.run(query, entity_id=entity_id)
return [{"path": str(record["path"]),
"distance": record["distance"]}
for record in result]
使用例
if __name__ == "__main__":
kg_client = KnowledgeGraphClient(
uri=os.getenv("NEO4J_URI"),
user=os.getenv("NEO4J_USER"),
password=os.getenv("NEO4J_PASSWORD")
)
# 顧客とイベントの記憶を追加
kg_client.add_memory_node("Customer", {
"id": "customer_001",
"name": "田中太郎",
"tier": "gold"
})
kg_client.add_memory_node("Event", {
"id": "event_001",
"type": "complaint",
"topic": "配送遅延"
})
kg_client.add_relationship(
"customer_001", "event_001",
"EXPERIENCED",
{"timestamp": "2025-01-15T10:30:00"}
)
print("ナレッジグラフに記憶を追加しました")
kg_client.close()Step 5：ハイブリッド記憶システムの統合
ここが核心です。ベクター検索とナレッジグラフ検索を融合し、スコアリングして最も関連性の高い記憶を返します。
import json
from typing import Optional
from embedder import HolySheepEmbedder
from knowledge_graph import KnowledgeGraphClient
import psycopg2
import numpy as np
class HybridMemorySystem:
"""ベクターDBとナレッジグラフのハイブリッド記憶システム"""
def __init__(self, db_config: dict, kg_client: KnowledgeGraphClient):
self.embedder = HolySheepEmbedder()
self.kg_client = kg_client
self.db_config = db_config
self.vector_weight = 0.6 # ベクター検索の重み
self.graph_weight = 0.4 # ナレッジグラフの重み
def _get_db_connection(self):
return psycopg2.connect(**self.db_config)
def store_memory(self, content: str, entity_id: str,
metadata: dict = None, related_entities: list[str] = None):
"""記憶を хранить (ベクター + ナレッジグラフ)"""
# 1. ベクターEmbeddingを生成
embedding = self.embedder.get_embedding(content)
# 2. PostgreSQLに保存
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO memories (entity_id, content, embedding, metadata)
VALUES (%s, %s, %s, %s)
RETURNING memory_id
""", (entity_id, content,
json.dumps(embedding),
json.dumps(metadata or {})))
memory_id = cur.fetchone()[0]
conn.commit()
finally:
conn.close()
# 3. ナレッジグラフにノード追加
self.kg_client.add_memory_node("Memory", {
"id": f"memory_{memory_id}",
"content": content,
"entity_id": entity_id
})
# 4. 関連エンティティとの関係を追加
if related_entities:
for related_id in related_entities:
self.kg_client.add_relationship(
entity_id, related_id, "RELATED_TO",
{"memory_id": memory_id}
)
return memory_id
def retrieve_memory(self, query: str, entity_id: Optional[str] = None,
top_k: int = 5) -> list[dict]:
"""ハイブリッド検索で記憶を取得"""
results = []
# 1. ベクター検索
query_embedding = self.embedder.get_embedding(query)
vector_results = self._vector_search(query_embedding, entity_id, top_k * 2)
# 2. ナレッジグラフ検索
if entity_id:
graph_results = self.kg_client.find_related_entities(entity_id, depth=3)
else:
graph_results = []
# 3. スコア融合
fused_results = self._fuse_scores(vector_results, graph_results, query_embedding)
return fused_results[:top_k]
def _vector_search(self, query_embedding: list[float],
entity_id: Optional[str], limit: int) -> list[dict]:
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
if entity_id:
cur.execute("""
SELECT memory_id, content, metadata,
1 - (embedding <=> %s) as similarity
FROM memories
WHERE entity_id = %s
ORDER BY embedding <=> %s
LIMIT %s
""", (json.dumps(query_embedding), entity_id,
json.dumps(query_embedding), limit))
else:
cur.execute("""
SELECT memory_id, content, metadata,
1 - (embedding <=> %s) as similarity
FROM memories
ORDER BY embedding <=> %s
LIMIT %s
""", (json.dumps(query_embedding),
json.dumps(query_embedding), limit))
return [{"memory_id": row[0], "content": row[1],
"metadata": json.loads(row[2]),
"vector_score": row[3]}
for row in cur.fetchall()]
finally:
conn.close()
def _fuse_scores(self, vector_results: list[dict],
graph_results: list[dict],
query_embedding: list[float]) -> list[dict]:
"""両方の検索結果を加重融合"""
score_map = {}
# ベクター結果を正規化スコアで追加
max_vector = max((r["vector_score"] for r in vector_results), default=1)
for r in vector_results:
norm_score = r["vector_score"] / max_vector
combined = norm_score * self.vector_weight
score_map[r["memory_id"]] = {
"content": r["content"],
"metadata": r["metadata"],
"combined_score": combined,
"vector_score": norm_score
}
# ナレッジグラフの結果をスコアに追加
for r in graph_results:
memory_ids = self._extract_memory_ids_from_path(r["path"])
distance_factor = 1 / (1 + r["distance"])
for mem_id in memory_ids:
if mem_id in score_map:
score_map[mem_id]["combined_score"] += (
distance_factor * self.graph_weight
)
score_map[mem_id]["graph_distance"] = r["distance"]
# スコア順でソート
sorted_results = sorted(
score_map.values(),
key=lambda x: x["combined_score"],
reverse=True
)
return sorted_results
@staticmethod
def _extract_memory_ids_from_path(path_str: str) -> list[str]:
"""パス文字列からmemory_IDを抽出"""
import re
return re.findall(r'memory_\d+', path_str)
使用例
if __name__ == "__main__":
# PostgreSQLテーブル作成（事前に実行が必要）
create_table_sql = """
CREATE TABLE IF NOT EXISTS memories (
memory_id SERIAL PRIMARY KEY,
entity_id VARCHAR(255) NOT NULL,
content TEXT NOT NULL,
embedding VECTOR(1536),
metadata JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_memories_embedding
ON memories USING ivfflat (embedding vector_cosine_ops);
CREATE INDEX IF NOT EXISTS idx_memories_entity
ON memories(entity_id);
"""
# ハイブリッド記憶システムの初期化
memory_system = HybridMemorySystem(
db_config={
"host": "localhost",
"port": 5432,
"database": "agent_memory",
"user": "your_user",
"password": "your_password"
},
kg_client=KnowledgeGraphClient(
os.getenv("NEO4J_URI"),
os.getenv("NEO4J_USER"),
os.getenv("NEO4J_PASSWORD")
)
)
# 記憶の保存
memory_id = memory_system.store_memory(
content="田中様は以前、配送遅延で苦情を経験された。対応には48時間要した。",
entity_id="customer_001",
metadata={"topic": "配送問題", "sentiment": "negative"},
related_entities=["product_001", "event_delivery_delay"]
)
print(f"記憶を保存しました: ID={memory_id}")
# 記憶の検索
results = memory_system.retrieve_memory(
query="過去に配送で問題があった顧客",
entity_id="customer_001",
top_k=3
)
for i, result in enumerate(results, 1):
print(f"\n--- 結果 {i} ---")
print(f"内容: {result['content']}")
print(f"総合スコア: {result['combined_score']:.3f}")
print(f"ベクタースコア: {result.get('vector_score', 0):.3f}")
memory_system.kg_client.close()Step 6：エージェントとの統合
実際にAIエージェントと統合して、長期記憶を活用した応答生成を行います。HolySheep AIのGPT-4o($8/MTok)またはコスト重視ならDeepSeek V3.2($0.42/MTok)が使えます。
import os
import requests
from dotenv import load_dotenv
load_dotenv()
class MemoryAwareAgent:
"""長期記憶を活用したAIエージェント"""
def __init__(self, memory_system: HybridMemorySystem):
self.memory_system = memory_system
self.api_key = os.getenv("HOLYSHEEP_API_KEY")
self.base_url = os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
self.model = "gpt-4o" # 高精度応答用
def chat(self, user_message: str, entity_id: str) -> str:
"""記憶を検索し、コンテキスト付きで応答生成"""
# 1. 関連記憶を検索
memories = self.memory_system.retrieve_memory(
query=user_message,
entity_id=entity_id,
top_k=5
)
# 2. 記憶をコンテキストに変換
context = self._build_context(memories)
# 3. プロンプト構築
system_prompt = f"""あなたは有用的な顧客サポートエージェントです。
以下の顧客履歴を考慮して、准确かつ丁寧な回答をしてください。
顧客履歴:
{context}
履歴がない場合は、一般的な知識で回答してください。"""
# 4. HolySheep AIで応答生成
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
"temperature": 0.7,
"max_tokens": 500
}
)
response.raise