AI 决策记忆不失效：基于 Harness 的受控记忆链路设计与实践

让 AI 记住历史决策，并不是把聊天记录全部塞回上下文这么简单。我在实际项目里遇到的一个典型问题是：同一个 AI 助手，昨天已经确定用 PostgreSQL 做用户数据存储，今天换了一个会话，它又开始认真比较 MySQL 和 MongoDB，完全无视已经形成的结论。后来我参考 DeepSeek Harness 这类方法的设计思路，把模型调用包装到一个受控的执行链路里：由代码负责记忆什么时候读、什么时候写，模型只负责根据给定上下文生成回答。这个改动之后，决策记忆才真正稳定下来。

这篇文章会从问题出发，先讲清楚决策记忆和普通对话记忆的差别，再给出一条可落地的决策记忆链路：消息入库、决策抽取、向量化、检索、Prompt 注入、模型调用、决策回写。文章最后会给出运行验证方式、常见排查链路和生产环境落地建议。即使你不使用 DeepSeek 系列模型，这套 Harness 思路也可以迁移到其他大模型应用里。

1. 先理解痛点：AI 会话的“失忆”为什么发生在决策层面

1.1 项目里的真实场景：换一个会话，结论就丢了

我参与的知识库助手项目里，AI 不只提供服务端问答，还会参与研发团队的技术讨论。产品经理和开发经常在同一个会话里讨论架构选型、排期、预算和工具选择。问题在于，这些结论散落在很长很长的对话流中，模型很难可靠地记住。

举例来说，某个周一的讨论里，团队明确决定“用户服务的数据存储使用 PostgreSQL，不再考虑 MongoDB”。这周之后，新开一个会话继续讨论监控方案，助手会把话题带回来，但如果直接问“数据库选型定了吗”，模型往往会给出两种数据库的对比，而不是直接说“已经定了，用 PostgreSQL”。原因不是模型不够聪明，而是上下文被其他对话冲淡，决策结论没有以结构化方式被记忆下来。

这类问题在普通聊天里很容易处理，但在工程场景里不能容忍。团队后面可能还会围绕这个决策继续做任务，如果 AI 每次重新推荐一个不同方案，讨论成本会成倍增加。

1.2 决策记忆与普通对话记忆的区别

决策记忆不是“把聊天记录保存下来”，也不是“把最近 N 条消息都带进 Prompt”。它要解决的是：在长会话和多会话场景中，快速、准确、按需地恢复那些已经形成的结论性信息。

下面这张表可以明确两者差异：

普通对话记忆解决的是“刚才聊到哪了”，决策记忆解决的是“这件事已经定了，后续不要再重新争论”。如果只是把对话历史全部塞进上下文，模型会同时看到“有人提过 MongoDB”和“团队最终选了 PostgreSQL”，它不一定知道哪个是最终结论。结构化决策库的职责就是把这层优先级用程序写清楚。

1.3 Harness 方法的核心：把记忆的读写权从模型手里拿回来

DeepSeek Harness 这类方法给我最大的启发，不是某个具体 API，而是“Harness”这个控制思想。Harness 可以理解为一个包裹模型调用的执行框架：模型不是自由直接回答用户，而是由外层代码先完成上下文检查、决策检索、条件校验，再把组织好的 Prompt 交给模型，最后对模型输出做解析和落库。

传统的构建方式里，开发者容易把“记忆”寄托给模型自身。比如把整段历史对话放在 Prompt 尾部，期望模型自动提取关键结论。这种做法在小样本场景有效，但一旦历史超过几千字，模型可能找不到关键决策，或者把讨论中的候选方案误认为最终结论。

Harness 方法改变的是控制权归属：

• 什么时候读记忆：由代码根据项目、会话、当前问题决定，不是把所有历史都塞进去。
• 读哪些记忆：由代码执行检索排序，不是按出现顺序全量拼接。
• 什么时候写记忆：由代码判断当前回复是否包含决策性内容，再触发结构化抽取。
• 写什么格式：由代码定义 JSON Schema，模型只是抽取器，不是数据库写入器。

这样一来，大模型被降级为“根据已确定上下文做推理”的组件，决策的保存和读取由稳定的程序逻辑控制。项目本身的稳定性不再完全依赖模型的临时发挥。

2. 设计决策记忆链路：一条从会话流水中抽取决策的流水线

2.1 整体数据流向

在实现之前，先明确整条链路的输入和输出。输入是一条新增消息，输出可能是普通回复，也可能是带决策约束的回复。中间经过哪些步骤，可以简化为下面的文本流程：

text复制用户消息
  -> 消息入库（去重、记录原始文本）
  -> 决策抽取（用模型输出结构化 JSON）
  -> JSON Schema 校验
  -> 决策落库
  -> 生成向量
  -> 下次提问时按语义检索历史决策
  -> 决策片段注入 Prompt
  -> 调用模型
  -> 回复用户
  -> 从模型输出中再次尝试抽取新决策（可选）

这个流程里，决策抽取和决策检索是两个最核心的环节。其他所有操作，比如建表、启动 API、调模型，都是为这两个环节服务。

2.2 决策事件的数据模型

决策要被检索、被审计、被更新，必须要有一个相对稳定的数据模型。我在项目里先建立了一张 decision_events 表，字段如下：

对应的 SQLite 建表语句可以这样写：

sql复制CREATE TABLE IF NOT EXISTS conversation_messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    message_hash TEXT NOT NULL UNIQUE,
    project_id TEXT NOT NULL,
    conversation_id TEXT NOT NULL,
    role TEXT NOT NULL,
    content TEXT NOT NULL,
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS decision_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id TEXT NOT NULL,
    conversation_id TEXT NOT NULL,
    source_message_id INTEGER NOT NULL,
    decision_type TEXT NOT NULL,
    title TEXT NOT NULL,
    decision_content TEXT NOT NULL,
    reason TEXT,
    alternative_rejected TEXT,
    importance_score REAL NOT NULL DEFAULT 1.0,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    FOREIGN KEY(source_message_id) REFERENCES conversation_messages(id)
);

这里有两个容易被忽略的设计点。第一，message_hash 要加唯一约束，目的是防止同一条消息被重复处理。第二，维度上先按 project_id 隔离，后期即使接入多个团队，也不会互相串决策。

向量可以单独放一张表，也可以直接作为 BLOB 存在 decision_events 里。小项目可以直接存 JSON 或 BLOB，先保证链路完整，后续再切换到正式向量数据库。

sql复制CREATE TABLE IF NOT EXISTS decision_embeddings (
    decision_id INTEGER PRIMARY KEY,
    embedding_model TEXT NOT NULL,
    embedding BLOB NOT NULL,
    updated_at TEXT NOT NULL
);

2.3 决策抽取的实现思路

决策抽取的目标是把一段对话文本转成上面这种结构化 JSON。实现上可以用大模型完成抽取，但外部代码必须做边界控制。

我在项目中定义了这样一个抽取函数：

python复制import json
import hashlib
from datetime import datetime


def make_message_hash(project_id: str, conversation_id: str, content: str) -> str:
    raw = f"{project_id}|{conversation_id}|{content}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def extract_decision(llm_client, message_text: str, conversation_context: str) -> dict:
    system_prompt = """
你是一个会话决策抽取器。你的任务是从对话中抽取“已经形成的结论性决策”。
普通闲聊、信息补充、候选方案讨论不算决策。
如果对话中没有形成决策，请严格返回：
{
  "has_decision": false
}
如果有决策，请返回如下 JSON：
{
  "has_decision": true,
  "decision_type": "architecture | budget | schedule | tool_selection | other",
  "title": "一句话概括决策",
  "decision_content": "明确的最终结论",
  "reason": "做这个决策的理由",
  "alternative_rejected": "被否决的方案或没有填 null",
  "importance_score": 1.0
}
注意：不要输出 JSON 以外的任何文字。
"""

    response = llm_client.chat.completions.create(
        model="your-model-name",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"对话背景：\n{conversation_context}\n\n最新消息：\n{message_text}"},
        ],
        temperature=0,
    )

    raw = response.choices[0].message.content.strip()
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        # 模型偶尔会输出 ```json 包裹，做一次剥离
        if raw.startswith("```"):
            raw = raw.strip("`")
            if raw.startswith("json"):
                raw = raw[4:].strip()
        data = json.loads(raw)

    if not data.get("has_decision"):
        return data

    required_keys = ["decision_type", "title", "decision_content", "reason", "alternative_rejected"]
    for key in required_keys:
        if key not in data:
            raise ValueError(f"decision extraction missing key: {key}")

    return data

这里的关键原因是：把 JSON 校验放在代码里，而不是让模型直接写数据库。模型输出的内容可能是乱格式，也可能是幻觉字段，只有代码层校验通过，才能进入决策库。这样即使模型换了版本，链路也不会跟着波动。

2.4 为什么抽取层要独立出来

新手容易把决策抽取逻辑写在聊天接口里，收到消息后当场调用模型抽取。这种做法不是不能用，但后续维护会很难。

独立抽取层有四个好处：

• 可以重放历史会话，批量生成决策库，不需要用户再次发言。
• 可以单独调节抽取 Prompt，不影响主对话质量。
• 可以给抽取请求加独立的超时、重试和日志，避免主接口被模型耗时拖垮。
• 可以接入多个入口，比如 Web 聊天、企微机器人、邮件归档，共用同一套决策库。

实际项目中，我会把抽取逻辑放到一个独立的 extractor 模块，主对话接口通过函数调用它，而不是让主对话接口承担所有事情。

3. 搭建最小可运行的决策记忆服务

3.1 环境准备与依赖

这个最小闭环可以全部跑在本地。需要满足以下几点：

对于本地演示，向量模型可以用 sentence-transformers 加载一个轻量模型。实际环境要按机器资源和模型兼容性来选择，不要照搬某个固定名字。下面示例只写调用方式，不绑定具体版本号。

安装示例：

bash复制pip install fastapi uvicorn sentence-transformers numpy openai

注意：这组依赖只适合本地跑通链路。生产环境要额外考虑连接池、模型网关限流、向量数据库和监控指标，不能直接用同一套配置上线。

3.2 项目目录结构

我建议按职责拆文件，避免把决策逻辑堆在一个 app.py 里。

text复制decision_harness/
├── api.py                 # FastAPI 接口
├── extractor.py           # 决策抽取
├── memory_manager.py      # 决策和消息落库
├── retriever.py           # 历史决策检索
├── prompt_builder.py      # 决策片段注入 Prompt
├── models.py              # 决策类型枚举
└── init.sql               # 建表 SQL

这六个文件对应一条清晰的处理链路。api.py 是入口，extractor.py 负责从消息中识别决策，memory_manager.py 负责存储，retriever.py 负责召回，prompt_builder.py 负责把召回结果变成可用的 Prompt 片段。

3.3 核心代码实现

先实现存储层。memory_manager.py 承担消息入库、决策入库、向量写入三件事：

python复制import sqlite3
import json
import numpy as np


class DecisionStore:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row

    def save_message(self, project_id: str, conversation_id: str, role: str, content: str):
        message_hash = make_message_hash(project_id, conversation_id, content)
        cur = self.conn.execute(
            """
            INSERT OR IGNORE INTO conversation_messages
            (message_hash, project_id, conversation_id, role, content, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (message_hash, project_id, conversation_id, role, content, datetime.utcnow().isoformat()),
        )
        self.conn.commit()
        row = self.conn.execute(
            "SELECT id FROM conversation_messages WHERE message_hash = ?",
            (message_hash,),
        ).fetchone()
        return row["id"]

    def save_decision(
        self,
        project_id: str,
        conversation_id: str,
        source_message_id: int,
        decision_type: str,
        title: str,
        decision_content: str,
        reason: str,
        alternative_rejected: str,
        importance_score: float,
    ) -> int:
        now = datetime.utcnow().isoformat()
        cur = self.conn.execute(
            """
            INSERT INTO decision_events
            (project_id, conversation_id, source_message_id, decision_type,
             title, decision_content, reason, alternative_rejected,
             importance_score, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                project_id, conversation_id, source_message_id, decision_type,
                title, decision_content, reason, alternative_rejected,
                importance_score, now, now,
            ),
        )
        self.conn.commit()
        return cur.lastrowid

    def save_embedding(self, decision_id: int, embedding_model: str, embedding):
        self.conn.execute(
            """
            INSERT OR REPLACE INTO decision_embeddings
            (decision_id, embedding_model, embedding, updated_at)
            VALUES (?, ?, ?, ?)
            """,
            (decision_id, embedding_model, np.asarray(embedding, dtype=np.float32).tobytes(), datetime.utcnow().isoformat()),
        )
        self.conn.commit()

代码里 INSERT OR IGNORE 配合消息哈希的唯一索引，解决了重复消息重复入库的问题。向量用 tobytes() 转成 BLOB，是为了在本地 SQLite 中省去额外的向量文件管理。

再实现检索层。retriever.py 负责把当前问题向量化，再和所有历史决策向量做相似度计算。示例代码故意保持简单，便于理解原理：

python复制import numpy as np
from datetime import datetime


def retrieve_decisions(store, embedder, project_id: str, query: str, top_k: int = 5):
    query_vec = embedder.encode(query, normalize_embeddings=True)

    rows = store.conn.execute(
        """
        SELECT d.id, d.title, d.decision_content, d.reason,
               d.alternative_rejected, d.importance_score, d.created_at,
               e.embedding
        FROM decision_events d
        LEFT JOIN decision_embeddings e ON e.decision_id = d.id
        WHERE d.project_id = ?
        ORDER BY d.created_at DESC
        """,
        (project_id,),
    ).fetchall()

    now = datetime.utcnow()
    scored = []

    for row in rows:
        if row["embedding"] is None:
            continue
        vec = np.frombuffer(row["embedding"], dtype=np.float32)
        similarity = float(np.dot(query_vec, vec))

        created_at = datetime.fromisoformat(row["created_at"])
        age_days = max((now - created_at).days, 0)

        # 时间衰减 + 重要性加权
        score = similarity / (1 + 0.05 * age_days) + 0.1 * row["importance_score"]
        scored.append((score, row))

    scored.sort(key=lambda x: x[0], reverse=True)
    return scored[:top_k]

向量相似度解决“语义接近”，但是工程上还要处理“新旧优先级”。两个决策都可能和当前问题语义接近，这时应当优先返回最新的那个。时间衰减项让旧决策不会永远压过新决策。

然后实现 Prompt 拼接。prompt_builder.py 把检索结果结构化输出：

python复制def build_decision_context(decisions) -> str:
    if not decisions:
        return ""

    lines = [
        "历史决策记录（以下决策优先级最高。除非用户明确表示要变更决策，否则回答必须与以下决策保持一致）："
    ]

    for _, d in decisions:
        rejected = d["alternative_rejected"] or "无"
        lines.append(
            f"- [{d['created_at']}] [{d['decision_type']}] "
            f"{d['title']}。结论：{d['decision_content']}。"
            f"理由：{d['reason'] or '未记录'}。"
            f"被否决方案：{rejected}。"
        )

    return "\n".join(lines)

为什么要把“被否决方案”也写进去？因为模型很容易把候选方案重新提出来。如果只写“选了 PostgreSQL”，模型后续不知道 MongoDB 已经被讨论过且被否掉，可能再次推荐 MongoDB。把否决项显式写进上下文，可以显著降低这种回退。

最后是 api.py，把完整链路串起来：

python复制from fastapi import FastAPI, Request
from extractor import extract_decision
from memory_manager import DecisionStore
from retriever import retrieve_decisions
from prompt_builder import build_decision_context

app = FastAPI()
store = DecisionStore("decision.db")
embedder = None  # 启动时加载，具体加载方式按模型来
llm_client = None


@app.post("/turns")
async def create_turn(req: Request):
    body = await req.json()
    project_id = body["project_id"]
    conversation_id = body["conversation_id"]
    user_message = body["message"]

    # 1. 保存用户消息
    source_msg_id = store.save_message(
        project_id=project_id,
        conversation_id=conversation_id,
        role="user",
        content=user_message,
    )

    # 2. 抽取当前消息中的决策
    decision_data = extract_decision(llm_client, user_message, "")

    # 3. 检索历史决策
    related = retrieve_decisions(
        store=store,
        embedder=embedder,
        project_id=project_id,
        query=user_message,
    )
    history_context = build_decision_context(related)

    # 4. 构建最终 Prompt
    final_prompt = f"{history_context}\n\n新消息：{user_message}"

    # 5. 调用模型
    response = llm_client.chat.completions.create(
        model="your-model-name",
        messages=[
            {"role": "system", "content": "你是一个研发协作助手，回答要尊重历史决策。"},
            {"role": "user", "content": final_prompt},
        ],
    )
    answer = response.choices[0].message.content

    # 6. 保存助手回复，为后续抽取提供上下文
    store.save_message(project_id, conversation_id, "assistant", answer)

    # 7. 如果抽取到了决策，落库
    if decision_data.get("has_decision"):
        decision_id = store.save_decision(
            project_id=project_id,
            conversation_id=conversation_id,
            source_message_id=source_msg_id,
            decision_type=decision_data["decision_type"],
            title=decision_data["title"],
            decision_content=decision_data["decision_content"],
            reason=decision_data.get("reason"),
            alternative_rejected=decision_data.get("alternative_rejected"),
            importance_score=decision_data.get("importance_score", 1.0),
        )
        # 生成向量，简单场景下可以直接同步处理
        embedding = embedder.encode(
            decision_data["decision_content"] + " " + decision_data.get("reason", ""),
            normalize_embeddings=True,
        )
        store.save_embedding(decision_id, "your-embedding-model", embedding)

    return {"answer": answer, "decision_recorded": decision_data.get("has_decision", False)}

3.4 启动和基础验证

本地启动：

bash复制uvicorn api:app --host 127.0.0.1 --port 8000

写入一条决策型消息：

bash复制curl -X POST http://127.0.0.1:8000/turns \
  -H "Content-Type: application/json" \
  -d '{
    "project_id": "demo",
    "conversation_id": "c1",
    "message": "用户服务数据库就用 PostgreSQL，不要再考虑 MongoDB 了。"
  }'

预期返回里的 decision_recorded 为 true。之后再次提问：

bash复制curl -X POST http://127.0.0.1:8000/turns \
  -H "Content-Type: application/json" \
  -d '{
    "project_id": "demo",
    "conversation_id": "c2",
    "message": "我们数据库选型定的是什么？"
  }'

预期输出应当提到 PostgreSQL，并避免推荐 MongoDB。这里的重点是验证“新会话”也能拿到旧会话里的决策。

4. 把历史决策注入新的对话 Prompt

4.1 注入内容和格式

决策注入不是简单把决策字段拼在消息前面，而是要把“优先级”表达清楚。下面是推荐片段格式：

text复制历史决策记录（以下决策优先级最高。除非用户明确表示要变更决策，否则回答必须与以下决策保持一致）：
- [2025-06-10T10:20:00] [tool_selection] 用户服务数据库选型。结论：使用 PostgreSQL。理由：团队已有运维经验，事务能力满足业务要求。被否决方案：MongoDB。

模型看到这样的内容后，理论上会把“PostgreSQL”当事实背景，而不是候选方案。但为了让效果更稳定，可以再给系统 Prompt 加一句约束：

text复制如果用户问题涉及历史决策中的内容，请先引用历史决策再回答。不要重新提出已被否决的方案。

4.2 检索策略：相似度、时间衰减和重要性加权

检索参数直接影响注入质量。下面是几个需要调优的参数：

调参时要通过一组固定问题来回归测试，不要凭感觉改。

如果提高 top_k，会发现 Prompt 越来越长，模型阅读成本增加。如果降低相似度阈值，又会把“讨论过但没有形成决定”的内容注入 Prompt，反而误导模型。时间衰减系数过大会让关键旧决策被漏掉，过小则会让每次新决策都很难覆盖旧决策。建议先固定 top_k 和阈值，只调时间衰减系数，观察一组测试问题的回答稳定性。

4.3 Harness 开始接管：完整请求链路

到这里，可以把整条链路写成一个受控的 handle_turn 函数，它就是项目里的 Harness：

python复制def handle_turn(project_id: str, conversation_id: str, user_message: str) -> dict:
    source_msg_id = store.save_message(project_id, conversation_id, "user", user_message)

    decision_data = extract_decision(llm_client, user_message, "")

    related = retrieve_decisions(store, embedder, project_id, user_message, top_k=5)
    history_context = build_decision_context(related)

    final_prompt = f"{history_context}\n\n新消息：{user_message}"
    answer = call_model(final_prompt)

    store.save_message(project_id, conversation_id, "assistant", answer)

    if decision_data.get("has_decision"):
        persist_decision(project_id, conversation_id, source_msg_id, decision_data)

    return {"answer": answer}

这个函数的意义在于，每次对话都被同一个控制流程包住。开发者可以在这里加日志、加鉴权、加超时、加一致性校验，而不需要改动业务代码。

实际项目中，还可以把 handle_turn 放到消息队列里异步执行。聊天接口先返回“处理中”，后台完成决策抽取和落库，避免模型调用耗时阻塞用户请求。

注意：Harness 不等于一个固定框架。它可以是一个类、一个装饰器、一个 Pipeline，核心是模型不能绕过这个链路直接访问数据库、直接修改记忆。

5. 运行验证：怎么确认 AI 真的记住了历史决策

5.1 构造一组测试数据

为了让验证可重复，先往决策库写入一组明确数据：

sql复制INSERT INTO decision_events (
    project_id, conversation_id, source_message_id, decision_type,
    title, decision_content, reason, alternative_rejected,
    importance_score, created_at, updated_at
) VALUES (
    'demo', 'c1', 1, 'tool_selection',
    '用户服务数据库选型',
    '使用 PostgreSQL',
    '团队已有运维经验，事务能力满足业务要求',
    'MongoDB',
    1.5,
    '2025-06-10T10:20:00',
    '2025-06-10T10:20:00'
);

再插入一条决策：日志系统使用 ELK，否决 Loki。这样测试时能观察检索是否区分不同领域。

5.2 对比有记忆和无记忆的模型输出

用同一句提问，分别比较两种模式：

注意，这里不是模型能力的测试，而是记忆链路有效性的测试。如果无记忆模式都能回答对，说明问题本身简单；只有无记忆时回答不稳定、有记忆时输出稳定的情况，才说明链路有效。

5.3 自动化检查清单

可以把验证过程写成断言脚本，集成到 CI 里：

python复制def check_decision_memory():
    # 1. 决策入库数量符合预期
    assert decision_count("demo") >= 2

    # 2. 向量检索能命中对应决策
    assert retrieve_top1("demo", "数据库选型")["title"] == "用户服务数据库选型"

    # 3. 构建出的 Prompt 包含历史决策片段
    context = build_decision_context(retrieve_decisions(...))
    assert "PostgreSQL" in context

    # 4. 模型回答不再重新推荐 MongoDB
    answer = call_model(context + "\n数据库选型定了吗？")
    assert "PostgreSQL" in answer

    # 5. 没有决策时，Prompt 不包含历史决策片段
    assert build_decision_context([]) == ""

这组断言不需要很复杂，但能防止后续改动破坏链路。尤其是修改了检索排序、Prompt 模板、抽取 Prompt 之后，跑一遍检查会非常省心。

6. 常见问题与排查链路

6.1 决策被重复入库

现象：同一条消息触发了多次请求，决策库里出现多条相同记录。

可能原因：消息去重失败；同一个 HTTP 请求被重试；抽取服务被并发调用。

检查方式：查 conversation_messages 中是否只有一条对应的原始消息，再查 decision_events 的 source_message_id 是否指向同一条消息。

sql复制SELECT source_message_id, COUNT(*)
FROM decision_events
GROUP BY source_message_id
HAVING COUNT(*) > 1;

解决方案：确认 message_hash 唯一索引已建立；在 DecisionStore.save_message 中使用 INSERT OR IGNORE；如果是异步任务，要保证同一个消息 ID 只有一个任务在处理，可以用 Redis 分布式锁或者数据库唯一约束。

6.2 检索不到刚写入的决策

现象：刚写入决策，立刻提问，模型回答里没有体现新决策。

可能原因：事务没有提交；向量没有生成成功；检索时相似度阈值太高；查询语句漏了 project_id 过滤。

检查方式：

sql复制SELECT * FROM decision_events ORDER BY id DESC LIMIT 5;
SELECT decision_id, length(embedding) FROM decision_embeddings ORDER BY updated_at DESC LIMIT 5;

如果 embedding 字段为空，说明向量写入失败。如果决策有记录但检索不到，先单独跑一次向量相似度计算，打印相似度分数。

解决方案：在写入决策后立即 commit；向量模型加载失败时要有明确日志；降低相似度阈值；确认检索 SQL 的 WHERE project_id = ? 是生效的。

注意：即使向量检索没有命中，也可以加一个“最近 24 小时决策兜底”的逻辑，确保刚写入的决策一定会出现在 Prompt 中，减少时序问题。

6.3 模型明明看到历史决策，仍然给出矛盾回答

现象：Prompt 里已经包含“使用 PostgreSQL，否决 MongoDB”，但模型回答还是要比较 MongoDB。

可能原因：历史决策片段被拼接在超过模型上下文容量的位置，模型没有真正“注意”到；系统 Prompt 没有强调决策优先级；当前用户消息语气过于强烈，模型跟着用户重新讨论已否决方案。

检查方式：打印最终 Prompt，人工确认决策片段是否完整；确认决策片段位置是不是在靠近输入末尾；查看系统 Prompt 中是否有“尊重历史决策”的约束。

解决方案：把决策片段放到 Prompt 的固定位置，建议放在 system 消息中；明确加一句“不得重新提出已否决方案”；在模型输出之后加一层后置校验，如果模型回答中重新出现了被否决方案，可以再次调用模型或直接提示用户“该方案已被历史决策否决”。

6.4 排错顺序清单

遇到决策记忆不生效时，按下面顺序排查，不要一上来就调模型参数：

1. 原始消息是否进入 conversation_messages。
2. 抽取层是否返回 has_decision=true。
3. 抽取 JSON 是否通过 Schema 校验。
4. 决策是否写入 decision_events。
5. 向量是否生成并写入 decision_embeddings。
6. 检索层是否返回该决策。
7. Prompt 中是否出现该决策片段。
8. 模型输出是否引用了该决策。

每一步都要有日志支撑。比如在抽取前后打两行日志，在检索后打印命中决策数量，在注入后打印 Prompt 前 500 字符。这样出现问题可以快速定位到具体环节。

7. 生产环境落地建议与扩展方向

7.1 学习环境与生产环境的差异

本地跑通的方案不能直接上线。至少需要区分下面这些差异：

生产环境里最容易被低估的是模型调用耗时。决策抽取和历史决策检索如果都放在用户请求链路里，一个请求可能要经过两到三次模型调用，整体延迟会很高。建议把“用户回复”和“决策汲取”拆开，用户先收到回复，后台再异步完成决策抽取和落库。

7.2 多项目隔离和隐私考虑

决策库里通常包含技术选型、业务规则、财务预算等敏感信息。检索时必须严格按 project_id 或 tenant_id 过滤，防止跨项目串数据。

几个实用建议：

• 所有查询语句强制带项目维度，不要提供不带 project_id 的查询函数。
• 日志中不要打印完整 decision_content，可以打印决策 ID 和标题。
• 重要项目的决策写入前要做人审或二次确认，不能只依赖模型抽取。
• 删除决策需要软删除，保留审计记录；不能直接物理删除后无法追溯。

如果团队规模不大，可以只做两级目录隔离：项目是最大隔离单位，会话是次一级隔离单位。检索时默认只查当前项目，不查历史全局。

7.3 从“记住决策”到“冲突检测和决策治理”

当前这套方案解决的是“记住结论”。再往前走，决策库还可以承担更主动的职责。

• 决策变更：当用户明确说“之前的决策改了”，代码需要支持把旧决策标记为失效，而不是继续让它参与检索。
• 冲突检测：新抽取的决策如果与旧决策矛盾，可以触发告警，由人工确认。
• 决策摘要归档：定期把某个项目下的所有决策汇总成文档，输出给团队查阅。
• 决策画像：统计哪些领域决策最多、哪类决策经常被推翻，辅助团队复盘。

这些方向并不需要全部第一期就做。建议先稳定住最小闭环：消息入库、决策抽取、向量检索、Prompt 注入、决策回写。跑通之后，再根据团队使用过程中出现的实际痛点逐项扩展。

7.4 最后的一点工程判断

让 AI 记住历史决策，核心不在于找到某个“记忆增强模型”，而在于把记忆当作系统的一部分来设计。模型负责理解语言和生成回答，代码负责决定记住什么、检索什么、注入什么。DeepSeek Harness 的方法给我的最大价值就是这个控制边界：不让模型临时决定一切，而是让工程链路稳定地管理记忆。

如果你想在真实项目中落地，建议先从一个项目、一条会话、两个关键决策开始。把决策抽取、向量化、检索和 Prompt 注入这条最小链路跑通，再逐渐加入变更、冲突检测、权限隔离和异步处理。这样每一步都有验证结果，不会因为模型输出的不确定性让整个系统看起来像黑盒。