memoryweave/task-phase1-1-models-queue.md at 995e93598aab46b51ac00c139a7a379e6519bec7

4.8 KiB

Raw Blame History

OpenCode 任务 - Phase 1.1 数据模型 + 持久化队列

日期: 2026-05-24 项目: 织忆 (MemoryWeave) 代码仓库: ~/projects/zhiyi/

背景

织忆是独立记忆服务，把对话日志蒸馏成结构化记忆。此任务是 Phase 1.1 的核心入口。

具体任务

任务 1: Episode 模型

文件: src/models/episode.py

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class Episode:
    """原始记忆单元 — 对话日志、任务记录等"""
    id: str                          # UUID
    timestamp: datetime              # 创建时间
    content: str                     # 原始内容
    entities: list[str] = field(default_factory=list)   # 实体列表
    facts: list[str] = field(default_factory=list)     # 事实列表
    metadata: dict = field(default_factory=dict)       # 元数据
    source: str = "hermes"           # 来源: hermes/openclaw/manual

要求:

dataclass 风格
有 to_dict() / from_dict() 序列化方法
UUID 生成用 uuid.uuid4()

任务 2: Distilled 模型

文件: src/models/distilled.py

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class Distilled:
    """蒸馏后的结构化记忆"""
    id: str
    episode_id: str                  # 来源 Episode ID
    type: str                        # "decision" | "request" | "fact" | "pattern"
    summary: str                     # 摘要
    entities: list[str] = field(default_factory=list)
    facts: list[str] = field(default_factory=list)
    confidence: float = 0.5          # 置信度 0-1
    status: str = "pending"          # "pending" | "validated" | "deprecated"
    importance: int = 0              # 重要性 0-5，>=3 永不衰减
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

要求:

同上，有序列化方法
status 可选值用常量或 enum

任务 3: 持久化队列

文件: src/distill/queue.py

import sqlite3
import json
from pathlib import Path
from datetime import datetime
from typing import Optional
from ..models.episode import Episode

class PersistenceQueue:
    """SQLite 持久化队列 — 入队/出队/持久化"""

    def __init__(self, db_path: str = "zhiyi.db"):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        """初始化表结构"""
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS episode_queue (
                id TEXT PRIMARY KEY,
                data TEXT NOT NULL,
                enqueued_at TEXT NOT NULL,
                dequeued_at TEXT,
                status TEXT DEFAULT 'pending'
            )
        """)
        self.conn.commit()

    def enqueue(self, episode: Episode) -> bool:
        """入队，返回是否成功"""

    def dequeue(self) -> Optional[Episode]:
        """出队，返回 Episode 或 None"""

    def peek(self) -> Optional[Episode]:
        """查看队首，不出队"""

    def size(self) -> int:
        """队列长度"""

    def is_empty(self) -> bool:
        """队列是否为空"""

    def requeue(self, episode: Episode) -> bool:
        """重新入队（处理失败时）"""

验收标准:

100条连续写入无丢失
服务重启后队列数据恢复
并发写入安全（加锁）

测试用例: tests/test_queue.py

def test_queue_persistence():
    q = PersistenceQueue(":memory:")  # 内存测试

    # 测试入队出队
    ep = Episode(id="1", timestamp=datetime.now(), content="test")
    q.enqueue(ep)
    assert q.size() == 1

    dequeued = q.dequeue()
    assert dequeued.id == "1"

    # 测试重连后恢复（内存队列不需要）
    # 真实 db 测试需要持久化路径

技术要求

语言: Python 3.10+
代码规范: PEP8
无外部依赖: 只用标准库 + sqlite3
测试覆盖: 每个文件有对应测试

参考

设计文档: ~/mc/小唯/07-Wiki/concepts/织忆(MemoryWeave)-v2.6-完整定稿.md 第4章
参考项目: ~/projects/memoryfabric-research/agent-memory-skill/memory-engine.py（线性衰减参考队列实现）

注意事项

不写设计之外的代码 — 按任务清单来
有问题先问 — 不要自作主张
完成后发飞书通知 — 牧尘或小唯
commit 要规范 — feat: Phase 1.1 数据模型 + 持久化队列

产出

文件	说明
src/models/episode.py	Episode 模型
src/models/distilled.py	Distilled 模型
src/distill/queue.py	持久化队列
tests/test_queue.py	队列测试
tests/test_models.py	模型测试

4.8 KiB Raw Blame History Unescape Escape