短期、中期、长期：一套生产级 Agent 记忆设计

2026-06-18 10:21:03 user: 这次别走原来的银行卡，我换了收款账户
2026-06-18 10:21:07 memory.write: append raw message, tokens=42
2026-06-18 10:27:19 user: 继续处理刚才那个
2026-06-18 10:27:22 memory.recall: top1="用户之前说不要走原来的银行卡"
2026-06-18 10:27:22 memory.recall: missing="新收款账户 account_id"
2026-06-18 10:27:24 tool.call: refund_retry(order_id=..., account_id=old_account)
2026-06-18 10:27:25 tool.result: rejected, reason=ACCOUNT_NOT_ALLOWED

def chat(user_id: str, message: str):
    history = db.load_messages(user_id)
    prompt = build_prompt(history, message)
    answer = llm.generate(prompt)
    db.save_message(user_id, "user", message)
    db.save_message(user_id, "assistant", answer)
    return answer

from dataclasses import dataclass, field

@dataclass
class AgentState:
    user_id: str
    thread_id: str
    goal: str | None = None
    messages: list[dict] = field(default_factory=list)
    tool_observations: list[dict] = field(default_factory=list)
    constraints: dict = field(default_factory=dict)

def build_short_context(state: AgentState, latest_user_msg: str) -> list[dict]:
    recent_messages = state.messages[-8:]
    tool_summary = summarize_tool_observations(state.tool_observations[-5:])

    return [
        {"role": "system", "content": "你是一个退款处理 Agent，必须遵守工具策略。"},
        {"role": "system", "content": f"当前目标：{state.goal or '待识别'}"},
        {"role": "system", "content": f"本轮工具观察：{tool_summary}"},
        *recent_messages,
        {"role": "user", "content": latest_user_msg},
    ]

{
  "summary": "用户咨询退款失败问题，后来想重新发起，期间提到更换收款账户。"
}

{
  "task_id": "refund_retry_20260618_1021",
"status": "waiting_user_confirm",
"order_id": "R202606180013",
"current_decision": "需要重新发起退款补偿",
"effective_constraints": {
    "payout_account": "new_account",
    "auto_submit": false
  },
"open_questions": [
    "等待用户确认新收款账户是否已完成风控校验"
  ],
"last_tool_errors": [
    {
      "tool": "refund_retry",
      "error": "ACCOUNT_NOT_ALLOWED",
      "time": "2026-06-18T10:27:25+08:00"
    }
  ]
}

def update_working_summary(old_summary: str, new_events: list[dict]) -> str:
    prompt = f"""
    你负责维护一个客服工单的工作摘要。
    只保留仍然影响后续执行的信息：
    1. 当前目标
    2. 已确认事实
    3. 有效约束
    4. 未解决问题
    5. 最近一次工具错误

    旧摘要：
    {old_summary}

    新事件：
    {new_events}

    输出 JSON，不要输出寒暄。
    """
    return llm.generate(prompt)

def apply_event(state: dict, event: dict) -> dict:
    if event["type"] == "user_constraint":
        key = event["key"]
        state["effective_constraints"][key] = {
            "value": event["value"],
            "source": event["message_id"],
            "updated_at": event["time"]
        }

    if event["type"] == "tool_error":
        state["last_tool_errors"] = [event]
        state["status"] = "blocked_by_tool_error"

    return state

state_freshness: 当前状态是否反映最新用户约束
state_coverage: 后续执行所需字段是否完整
conflict_count: 新旧事实冲突有没有被识别
resume_success_rate: 隔一段时间恢复任务，Agent 是否能接着做

namespace = ["user", user_id, "profile"]
key       = "communication_preferences"

namespace = ["org", org_id, "refund_agent", "rules"]
key       = "retry_policy"

namespace = ["project", repo_id, "lessons"]
key       = "release_checklist"

def maybe_write_long_term_memory(event: dict, state: dict):
    candidate = extract_memory_candidate(event, state)
    if candidate isNone:
        return

    decision = classify(candidate)

    if decision.kind notin {"preference", "stable_fact", "lesson", "policy"}:
        return

    if decision.confidence < 0.85:
        queue_for_review(candidate)
        return

    old = memory_store.get(namespace=decision.namespace, key=decision.key)
    merged = merge_with_conflict_check(old, candidate)

    if merged.has_conflict:
        ask_user_or_human_reviewer(merged)
        return

    memory_store.put(
        namespace=decision.namespace,
        key=decision.key,
        value=merged.value,
        metadata={
            "source_message_id": event["message_id"],
            "confidence": decision.confidence,
            "ttl_days": decision.ttl_days,
            "updated_at": event["time"]
        }
    )

memories = vector_store.search(user_message, top_k=8)
prompt = build_prompt(history, memories, user_message)

用户原话：
继续处理刚才那个

改写后的检索问题：
继续处理订单 R202606180013 的退款补偿任务。
需要考虑用户已要求使用新收款账户，且上一次 refund_retry 工具返回 ACCOUNT_NOT_ALLOWED。

def recall_memory(user_id: str, org_id: str, state: dict, message: str):
    query = rewrite_query(state, message)

    profile = profile_store.get(["user", user_id, "profile"])
    task_state = task_store.get(state["task_id"])
    lessons = vector_store.search(
        namespace=["org", org_id, "refund_agent", "lessons"],
        query=query,
        top_k=5
    )
    policies = rule_store.match(
        namespace=["org", org_id, "refund_agent", "rules"],
        facts=task_state
    )

    return compose_context(profile, task_state, lessons, policies)

<current_task_state>
order_id=R202606180013
status=waiting_user_confirm
effective_constraints.payout_account=new_account
</current_task_state>

<user_preferences>
回复时先给结论，再给操作明细。
</user_preferences>

<org_policies>
当 refund_retry 返回 ACCOUNT_NOT_ALLOWED 时，禁止自动重试，必须升级人工校验账户。
</org_policies>

<retrieved_lessons>
历史经验：收款账户变更后，必须先调用 account_risk_check。
</retrieved_lessons>

User Message
    |
    v
Short-term State
    - recent messages
    - current goal
    - tool observations
    |
    v
Mid-term Task Memory
    - task summary
    - confirmed facts
    - effective constraints
    - open questions
    |
    v
Long-term Memory Recall
    - user profile
    - org policies
    - project lessons
    - semantic search results
    |
    v
Context Composer -> LLM -> Tool / Answer

LLM / Tool / User Event
    |
    v
Memory Candidate Extractor
    |
    v
Classifier
    - short-term only
    - update task state
    - write long-term memory
    - ignore
    |
    v
Conflict Check / TTL / Approval
    |
    v
Memory Stores

case_01 短期指代：
用户第 1 轮给订单号，第 3 轮说“继续处理这个”，Agent 能否补全 order_id。

case_02 中期恢复：
会话中断 30 分钟后恢复，Agent 能否知道当前状态、未完成问题、最新约束。

case_03 长期偏好：
用户上周明确说“所有执行前先预览”，今天 Agent 是否仍然遵守。

case_04 冲突覆盖：
用户先说用账户 A，后说改成账户 B，Agent 是否只使用最新有效账户。

case_05 过期记忆：
用户说“今天临时不要自动提交”，第二天是否不再错误召回。

case_06 权限隔离：
用户 A 的项目经验，是否不会泄漏给用户 B。

用 Redis 存短期记忆，用向量库存长期记忆。

我会把 Agent 记忆拆成短期、中期、长期三层。

短期记忆是当前 thread/run 的上下文，包括最近消息、当前目标、工具观察和临时约束，直接参与本轮 prompt 组装。

中期记忆是当前会话或任务的工作状态，不保存完整流水账，而是维护结构化摘要，比如已确认事实、有效约束、未解决问题和最后一次工具错误。

长期记忆是跨会话复用的信息，比如用户偏好、组织规则、项目经验和历史教训。它不能简单每轮写入，需要候选提取、分类、置信度、TTL、来源追踪和冲突校验。

召回时也不能只靠向量库，要先结合当前任务做 query rewrite，再按 profile、task state、policy、lesson 分源召回，最后由 context composer 按优先级放回模型上下文。

我会重点关注 stale memory、冲突覆盖、权限隔离和 resume success rate，因为这些才是生产里最容易出问题的地方。