多轮对话上下文

要点

LLM 本身没有记忆，每次调用都是独立的
多轮对话需要应用层维护完整的消息历史
上下文窗口有限，需要策略性地管理历史消息
常见的管理策略：滑动窗口、摘要压缩、向量检索
上下文管理影响对话质量和成本
AI 项目需要平衡上下文长度和信息完整性

1. 为什么需要上下文管理

1.1 LLM 的局限

LLM 没有记忆，每次调用都是独立的：

// 第一次调用
await callLLM('我叫 Alice')
// AI: 你好 Alice！
 
// 第二次调用（新的请求）
await callLLM('我叫什么名字？')
// AI: 我不知道你叫什么名字。

1.2 解决方案

把之前的对话历史一起发送给 LLM：

// 发送完整的对话历史
await callLLM([
  { role: 'user', content: '我叫 Alice' },
  { role: 'assistant', content: '你好 Alice！' },
  { role: 'user', content: '我叫什么名字？' },
])
// AI: 你叫 Alice。

2. 上下文窗口限制

2.1 各模型的上下文窗口

模型	上下文窗口
GPT-4	8K / 32K / 128K tokens
GPT-3.5	4K / 16K tokens
Claude 3	200K tokens
Gemini 1.5	1M tokens

2.2 Token 计算

import { encoding_for_model } from 'tiktoken'
 
const encoder = encoding_for_model('gpt-4')
 
function countTokens(text: string): number {
  return encoder.encode(text).length
}
 
// 示例
const text = '这是一段中文文本'
const tokens = countTokens(text)
console.log(tokens)  // 大约 10-15 个 token

2.3 成本问题

上下文越长，成本越高：

// 假设 GPT-4 价格：$0.03 / 1K input tokens
 
// 短上下文（1000 tokens）
const cost1 = 1000 / 1000 * 0.03  // $0.03
 
// 长上下文（100000 tokens）
const cost2 = 100000 / 1000 * 0.03  // $3.00
 
// 100 倍的成本差异！

3. 上下文管理策略

3.1 滑动窗口

只保留最近 N 轮对话：

function slidingWindow(messages: Message[], maxTurns: number = 10): Message[] {
  const system = messages.filter(m => m.role === 'system')
  const conversation = messages.filter(m => m.role !== 'system')
 
  // 每轮 2 条消息（user + assistant）
  const recent = conversation.slice(-maxTurns * 2)
 
  return [...system, ...recent]
}
 
// 使用
const messages = [
  { role: 'system', content: '你是一位编程导师。' },
  // ... 50 轮对话 ...
]
 
const truncated = slidingWindow(messages, 10)  // 只保留最近 10 轮

优点：简单、可控成本缺点：丢失早期信息

3.2 Token 限制

精确控制总 token 数：

function truncateByTokens(
  messages: Message[],
  maxTokens: number
): Message[] {
  const system = messages.filter(m => m.role === 'system')
  const conversation = messages.filter(m => m.role !== 'system')
 
  let totalTokens = system.reduce(
    (sum, m) => sum + countTokens(m.content as string),
    0
  )
 
  const result = [...system]
 
  // 从最新的消息开始添加
  for (let i = conversation.length - 1; i >= 0; i--) {
    const msg = conversation[i]
    const tokens = countTokens(msg.content as string)
 
    if (totalTokens + tokens > maxTokens) {
      break
    }
 
    result.splice(1, 0, msg)
    totalTokens += tokens
  }
 
  return result
}
 
// 使用
const truncated = truncateByTokens(messages, 4000)  // 限制 4000 tokens

优点：精确控制成本缺点：可能截断重要信息

3.3 摘要压缩

对旧对话进行摘要：

async function compressHistory(messages: Message[]): Promise<Message[]> {
  if (messages.length <= 20) {
    return messages  // 消息不多，不需要压缩
  }
 
  const system = messages.filter(m => m.role === 'system')
  const recent = messages.slice(-10)  // 保留最近 5 轮
  const old = messages.filter(m => m.role !== 'system').slice(0, -10)
 
  // 对旧消息进行摘要
  const summaryPrompt = `
将以下对话总结为简短的摘要，保留关键信息：
 
${old.map(m => `${m.role}: ${m.content}`).join('\n')}
 
请用 2-3 句话总结。
`
 
  const response = await callLLM(summaryPrompt)
  const summary = response.choices[0].message.content
 
  return [
    ...system,
    {
      role: 'system',
      content: `之前对话的摘要：${summary}`
    },
    ...recent,
  ]
}

优点：保留关键信息，控制长度缺点：摘要可能丢失细节，需要额外的 LLM 调用

3.4 向量检索

从历史对话中检索相关内容：

// 预先存储消息的向量
async function storeMessage(message: Message) {
  const embedding = await getEmbedding(message.content as string)
 
  await vectorDb.insert({
    messageId: message.id,
    embedding,
    content: message.content,
  })
}
 
// 查询时检索相关消息
async function retrieveRelevantMessages(
  query: string,
  limit: number = 10
): Promise<Message[]> {
  const queryEmbedding = await getEmbedding(query)
 
  const results = await vectorDb.query({
    vector: queryEmbedding,
    limit,
  })
 
  return results.map(r => ({
    role: 'user',  // 或从数据库查询
    content: r.content,
  }))
}
 
// 构建上下文
async function buildContext(query: string): Promise<Message[]> {
  const relevantMessages = await retrieveRelevantMessages(query)
 
  return [
    { role: 'system', content: '根据之前的对话回答问题。' },
    {
      role: 'system',
      content: `相关的历史对话：\n${relevantMessages.map(m => m.content).join('\n')}`
    },
    { role: 'user', content: query },
  ]
}

优点：保留相关信息，不受时间顺序限制缺点：需要向量数据库，检索可能不准确

3.5 混合策略

结合多种策略：

async function buildMessages(
  systemPrompt: string,
  history: Message[],
  currentMessage: string,
  maxTokens: number
): Promise<Message[]> {
  // 1. 始终保留 system prompt
  const messages: Message[] = [
    { role: 'system', content: systemPrompt }
  ]
 
  // 2. 检索相关的历史消息
  const relevant = await retrieveRelevantMessages(currentMessage, 5)
  messages.push(...relevant)
 
  // 3. 添加最近的对话
  const recent = history.slice(-10)
  messages.push(...recent)
 
  // 4. 添加当前消息
  messages.push({ role: 'user', content: currentMessage })
 
  // 5. 如果超过 token 限制，进行压缩
  const totalTokens = messages.reduce(
    (sum, m) => sum + countTokens(m.content as string),
    0
  )
 
  if (totalTokens > maxTokens) {
    return await compressHistory(messages)
  }
 
  return messages
}

4. 完整实现示例

4.1 数据库设计

export const conversations = pgTable('conversations', {
  id: uuid('id').primaryKey().defaultRandom(),
  userId: uuid('user_id').notNull(),
  title: text('title'),
  createdAt: timestamp('created_at').defaultNow().notNull(),
})
 
export const messages = pgTable('messages', {
  id: uuid('id').primaryKey().defaultRandom(),
  conversationId: uuid('conversation_id').notNull(),
  role: text('role').notNull(),
  content: text('content').notNull(),
  tokenCount: integer('token_count'),
  createdAt: timestamp('created_at').defaultNow().notNull(),
})
 
// 消息向量（用于检索）
export const messageEmbeddings = pgTable('message_embeddings', {
  id: uuid('id').primaryKey().defaultRandom(),
  messageId: uuid('message_id').notNull(),
  embedding: vector('embedding', { dimensions: 1536 }),
})

4.2 对话服务

class ConversationService {
  async chat(conversationId: string, userMessage: string) {
    // 1. 获取对话历史
    const conversation = await db.query.conversations.findFirst({
      where: eq(conversations.id, conversationId),
      with: {
        messages: {
          orderBy: asc(messages.createdAt),
        },
      },
    })
 
    // 2. 构建消息列表
    const messages = await this.buildMessages(
      conversation.messages,
      userMessage
    )
 
    // 3. 调用 LLM
    const response = await callLLM(messages)
    const reply = response.choices[0].message.content
 
    // 4. 保存消息
    await db.insert(messages).values([
      {
        conversationId,
        role: 'user',
        content: userMessage,
        tokenCount: countTokens(userMessage),
      },
      {
        conversationId,
        role: 'assistant',
        content: reply,
        tokenCount: countTokens(reply),
      },
    ])
 
    // 5. 存储向量（异步）
    this.storeEmbedding(conversationId, userMessage).catch(console.error)
 
    return reply
  }
 
  private async buildMessages(
    history: Message[],
    currentMessage: string
  ): Promise<Message[]> {
    const system = { role: 'system', content: '你是一位友好的编程导师。' }
 
    // 策略 1: 滑动窗口
    const recentHistory = history.slice(-20)  // 最近 10 轮
 
    // 策略 2: 向量检索（可选）
    const relevant = await this.retrieveRelevant(currentMessage)
 
    // 组合
    const messages = [
      system,
      ...relevant,
      ...recentHistory,
      { role: 'user', content: currentMessage },
    ]
 
    // 策略 3: Token 限制
    return this.truncateByTokens(messages, 4000)
  }
 
  private async retrieveRelevant(query: string): Promise<Message[]> {
    const queryEmbedding = await getEmbedding(query)
 
    const results = await db.execute(sql`
      SELECT m.content, m.role
      FROM message_embeddings e
      JOIN messages m ON m.id = e.message_id
      ORDER BY e.embedding <=> ${queryEmbedding}::vector
      LIMIT 5
    `)
 
    return results.rows.map((r: any) => ({
      role: r.role,
      content: r.content,
    }))
  }
 
  private truncateByTokens(messages: Message[], maxTokens: number): Message[] {
    let totalTokens = 0
    const result: Message[] = []
 
    // 从后往前添加（最新的消息优先）
    for (let i = messages.length - 1; i >= 0; i--) {
      const msg = messages[i]
      const tokens = countTokens(msg.content as string)
 
      if (totalTokens + tokens > maxTokens) {
        break
      }
 
      result.unshift(msg)
      totalTokens += tokens
    }
 
    return result
  }
 
  private async storeEmbedding(conversationId: string, content: string) {
    const embedding = await getEmbedding(content)
 
    await db.insert(messageEmbeddings).values({
      messageId: conversationId,  // 简化示例
      embedding,
    })
  }
}

5. 最佳实践

5.1 保留 System Prompt

System prompt 对 AI 行为很重要，应该始终保留：

// ✅ 保留 system prompt
const messages = [
  { role: 'system', content: systemPrompt },
  ...truncatedHistory,
]
 
// ❌ 不要截断 system prompt
const messages = truncatedHistory  // 可能丢失 system prompt

5.2 优先保留最新消息

最近的对话通常更重要：

// ✅ 从最新消息开始添加
for (let i = history.length - 1; i >= 0; i--) {
  if (totalTokens + tokens > maxTokens) break
  result.unshift(history[i])
}
 
// ❌ 从最旧的消息开始
for (const msg of history) {
  if (totalTokens + tokens > maxTokens) break
  result.push(msg)
}

5.3 监控 Token 使用

async function trackTokenUsage(userId: string, tokens: number) {
  await db.insert(tokenUsage).values({
    userId,
    tokens,
    createdAt: new Date(),
  })
 
  // 检查是否接近限额
  const totalUsage = await getTotalUsage(userId)
  if (totalUsage > QUOTA_LIMIT * 0.8) {
    await sendWarning(userId, 'Token usage approaching limit')
  }
}

5.4 缓存常用上下文

const contextCache = new Map<string, Message[]>()
 
async function getCachedContext(conversationId: string): Promise<Message[]> {
  const cached = contextCache.get(conversationId)
  if (cached) return cached
 
  const messages = await loadMessages(conversationId)
  contextCache.set(conversationId, messages)
 
  // 5 分钟后清除缓存
  setTimeout(() => contextCache.delete(conversationId), 5 * 60 * 1000)
 
  return messages
}

总结

多轮对话上下文管理是 AI 应用的核心挑战。需要在信息完整性、成本、性能之间找到平衡。

这一节涉及到的几个实践：

上下文窗口：不同模型有不同的限制
滑动窗口：保留最近 N 轮对话
Token 限制：精确控制总 token 数
摘要压缩：对旧对话进行摘要
向量检索：检索相关的历史消息
混合策略：结合多种方法

上下文管理没有银弹，需要根据具体场景选择合适的策略。关键是要监控 token 使用，控制成本，同时保持对话质量。

下一篇看 Token 概念与成本——如何计算和控制 AI API 的成本。