用 Redis 承载 Agent 会话记忆：存储边界、压缩时机与并发写入

// Node 18+ / ioredis 5.x
import Redis from 'ioredis'

export const redis = new Redis({
  host: process.env.REDIS_HOST ?? '127.0.0.1',
  port: Number(process.env.REDIS_PORT ?? 6379),
  password: process.env.REDIS_PASSWORD || undefined,
  db: 0,
  maxRetriesPerRequest: 3,
  enableReadyCheck: true,
})

redis.on('error', (err) => {
  // 必须监听：未处理的 error 事件会让 Node 进程直接退出
  console.error('[redis] connection error', err)
})

import { HumanMessage, AIMessage, SystemMessage, ToolMessage } from '@langchain/core/messages'

const MESSAGE_CTOR = {
  human: HumanMessage,
  ai: AIMessage,
  system: SystemMessage,
  tool: ToolMessage,
}

function toStored(messages) {
  return messages.map((m) => {
    const type = m._getType()
    const data = { content: m.content }
    if (m.additional_kwargs && Object.keys(m.additional_kwargs).length) {
      data.additional_kwargs = m.additional_kwargs
    }
    if (m.tool_calls?.length) data.tool_calls = m.tool_calls
    if (m.tool_call_id) data.tool_call_id = m.tool_call_id
    if (m.name) data.name = m.name
    if (m.id) data.id = m.id
    return { type, data }
  })
}

function fromStored(stored) {
  return stored.map(({ type, data }) => {
    if (type === 'tool') {
      return new ToolMessage(data)
    }
    const Ctor = MESSAGE_CTOR[type]
    if (!Ctor) throw new Error(`unknown message type: ${type}`)
    return new Ctor(data)
  })
}

export class RedisMessageStore {
  #redis
  #prefix
  #ttlSeconds

  constructor({ redis, prefix = 'agent:short_memory:', ttlSeconds = 1800 }) {
    this.#redis = redis
    this.#prefix = prefix
    this.#ttlSeconds = ttlSeconds
  }

  #key(sessionId) {
    return `${this.#prefix}{${sessionId}}:messages`
  }

  async load(sessionId) {
    const raw = await this.#redis.get(this.#key(sessionId))
    if (raw === null) return []
    return fromStored(JSON.parse(raw))
  }

  async save(sessionId, messages) {
    const payload = JSON.stringify(toStored(messages))
    await this.#redis.set(this.#key(sessionId), payload, 'EX', this.#ttlSeconds)
  }

  async ttl(sessionId) {
    return this.#redis.ttl(this.#key(sessionId))
  }
}

import { HumanMessage, SystemMessage } from '@langchain/core/messages'

async function compressIfNeeded(messages, summaryModel, { maxMessages = 8, keep = 4 } = {}) {
  if (messages.length <= maxMessages) return messages

  const older = messages.slice(0, -keep)
  const recent = messages.slice(-keep)
  const transcript = older
    .map((m) => `${m._getType()}: ${typeof m.content === 'string' ? m.content : JSON.stringify(m.content)}`)
    .join('\n')

  const summary = await summaryModel.invoke([
    new SystemMessage('把下面的对话压缩成要点，保留事实、数字、承诺和未完成事项，不要评论。'),
    new HumanMessage(transcript),
  ])

  return [new SystemMessage(`更早对话的摘要：\n${summary.content}`), ...recent]
}

import { randomUUID } from 'node:crypto'

const RELEASE = `if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('del', KEYS[1])
else
  return 0
end`

const sleep = (ms) => new Promise((r) => setTimeout(r, ms))

export class SessionLock {
  #redis
  #ttlMs
  #waitMs

  constructor(redis, { ttlMs = 15000, waitMs = 3000 } = {}) {
    this.#redis = redis
    this.#ttlMs = ttlMs
    this.#waitMs = waitMs
  }

  async run(sessionId, fn) {
    const key = `agent:lock:{${sessionId}}`
    const token = randomUUID()
    const deadline = Date.now() + this.#waitMs

    let acquired = false
    while (!acquired && Date.now() <= deadline) {
      acquired = (await this.#redis.set(key, token, 'PX', this.#ttlMs, 'NX')) === 'OK'
      if (!acquired) await sleep(50 + Math.floor(Math.random() * 50))
    }
    if (!acquired) throw new Error(`session busy: ${sessionId}`)

    try {
      return await fn()
    } finally {
      // 只删自己持有的锁，避免误删别人的
      await this.#redis.eval(RELEASE, 1, key, token)
    }
  }
}

agent:short_memory:{sessionId}:messages   String，JSON 消息数组
agent:short_memory:{sessionId}:meta       Hash，用户信息、轮次、创建时间
agent:long_memory:{userId}                长期记忆，实际走向量库