Weaviate Engram 深度解析：AI Agent 生产级记忆服务的技术架构与工程实践

# Weaviate Engram Python SDK 示例
import weaviate
from weaviate.classes.init import Auth

# 连接 Weaviate Cloud
client = weaviate.connect_to_weaviate_cloud(
    cluster_url="https://your-cluster.weaviate.network",
    auth_credentials=Auth.api_key("your-api-key"),
)

# 使用 Engram 记忆服务
from weaviate.engram import EngramClient

engram = EngramClient(client)

# 1. 摄入记忆（异步 Pipeline）
engram.ingest(
    project="my-app",
    user="user-123",
    property="preferences",
    events=[
        {
            "type": "conversation",
            "content": "用户偏好中文回复，不喜欢过于正式的语气",
            "metadata": {"source": "chat", "timestamp": "2026-06-17T10:00:00Z"}
        }
    ]
)

# 2. 检索记忆（混合检索）
memories = engram.retrieve(
    project="my-app",
    user="user-123",
    query="用户喜欢什么语言？",
    properties=["preferences"],  # 只检索偏好分区
    limit=5,
    search_type="hybrid"  # 向量 + BM25 混合
)

for mem in memories:
    print(f"[{mem.property}] {mem.content} (score: {mem.score:.3f})")
    # 输出: [preferences] 用户偏好中文回复... (score: 0.92)

// Weaviate Engram TypeScript SDK 示例
import { WeaviateClient, EngramClient } from 'weaviate-client';

const client = new WeaviateClient({
  scheme: 'https',
  host: 'your-cluster.weaviate.network',
  apiKey: 'your-api-key',
});

const engram = new EngramClient(client);

// 摄入记忆
await engram.ingest({
  project: 'my-app',
  user: 'user-123',
  property: 'preferences',
  events: [
    {
      type: 'conversation',
      content: '用户偏好中文回复，不喜欢过于正式的语气',
      metadata: { source: 'chat', timestamp: new Date().toISOString() },
    },
  ],
});

// 检索记忆
const memories = await engram.retrieve({
  project: 'my-app',
  user: 'user-123',
  query: '用户喜欢什么语言？',
  properties: ['preferences'],
  limit: 5,
  searchType: 'hybrid',
});

memories.forEach((mem) => {
  console.log(`[${mem.property}] ${mem.content} (score: ${mem.score.toFixed(3)})`);
});

# 多 Agent 系统的 Engram 集成示例
from weaviate.engram import EngramClient

class MultiAgentMemory:
    """多 Agent 共享记忆管理器"""

    def __init__(self, engram: EngramClient):
        self.engram = engram
        self.project = "customer-service"

    async def store_user_profile(self, user_id: str, profile: dict):
        """存储用户基本信息（所有 Agent 共享）"""
        await self.engram.ingest(
            project=self.project,
            user=user_id,
            property="profile",
            events=[{
                "type": "profile_update",
                "content": str(profile),
                "metadata": {"updated_at": "2026-06-17T10:00:00Z"}
            }]
        )

    async def store_agent_interaction(
        self, user_id: str, agent_type: str, interaction: str
    ):
        """存储 Agent 交互历史（按 Agent 类型分区）"""
        property_map = {
            "tech": "tech-history",
            "biz": "biz-history",
        }
        await self.engram.ingest(
            project=self.project,
            user=user_id,
            property=property_map[agent_type],
            events=[{
                "type": "interaction",
                "content": interaction,
                "metadata": {"agent": agent_type}
            }]
        )

    async def get_context_for_agent(
        self, user_id: str, agent_type: str, query: str
    ) -> str:
        """为特定 Agent 获取上下文"""
        # 始终加载共享记忆
        shared = await self.engram.retrieve(
            project=self.project,
            user=user_id,
            query=query,
            properties=["profile", "preferences"],
            limit=3,
        )

        # 加载 Agent 专有记忆
        property_map = {
            "tech": "tech-history",
            "biz": "biz-history",
        }
        specific = await self.engram.retrieve(
            project=self.project,
            user=user_id,
            query=query,
            properties=[property_map[agent_type]],
            limit=5,
        )

        # 组合上下文
        context_parts = []
        context_parts.append("=== 用户信息 ===")
        for mem in shared:
            context_parts.append(f"[{mem.property}] {mem.content}")
        context_parts.append(f"\n=== {agent_type} Agent 历史 ===")
        for mem in specific:
            context_parts.append(f"[{mem.property}] {mem.content}")

        return "\n".join(context_parts)


# 使用示例
async def main():
    memory = MultiAgentMemory(engram_client)

    # 路由 Agent 分发技术问题到技术 Agent
    context = await memory.get_context_for_agent(
        user_id="user-123",
        agent_type="tech",
        query="用户的服务器部署问题"
    )
    print(context)