徐涛
6e1182e64c
为 v0.2.0-rc.1 API 稳定性收尾。覆盖： - P0 核心 IR: Message / ContentBlock / ContentBlockType / StreamEvent / HookEvent - P0 Error: AgentError / LlmError / ToolError / MemoryError / PromptError - P1 其他: MemoryStrategy / StepStatus / ToolChoice / ResponseFormat 明确不加的: 内部 wire-format (OpenaiChatMessage 等) / 语义已收敛 (Role/ServiceTier 等) / 使用面窄 (Permission 等)。 示例侧的 3 处 Message exhaustive match (prompt_composer / conversation_memory_demo) 补全 `_` 通配分支，零行为变化。 验收: cargo build + clippy -D warnings + test --all-targets 全绿 (200 passed)
280 lines
9.2 KiB
Rust
280 lines
9.2 KiB
Rust
//! 对话记忆 —— 多轮对话消息管理，复用 `llm::compact` 的压缩逻辑。
|
||
|
||
use std::sync::Arc;
|
||
|
||
use time::OffsetDateTime;
|
||
|
||
use crate::llm::compact::{CompactConfig, CompactState, microcompact, should_compact};
|
||
use crate::llm::types::message::Message;
|
||
use crate::memory::error::MemoryError;
|
||
use crate::memory::store::MemoryStore;
|
||
use crate::memory::types::MemoryItem;
|
||
|
||
/// 对话消息管理策略。
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||
#[non_exhaustive]
|
||
pub enum MemoryStrategy {
|
||
/// 滑动窗口：达到上限时删除最旧消息。
|
||
SlidingWindow,
|
||
/// 保留所有消息（仅压缩，不删除）。
|
||
Full,
|
||
}
|
||
|
||
/// 对话记忆配置。
|
||
#[derive(Debug, Clone)]
|
||
pub struct ConversationMemoryConfig {
|
||
pub strategy: MemoryStrategy,
|
||
pub max_turns: usize,
|
||
pub compact_config: Option<CompactConfig>,
|
||
}
|
||
|
||
impl Default for ConversationMemoryConfig {
|
||
fn default() -> Self {
|
||
Self {
|
||
strategy: MemoryStrategy::SlidingWindow,
|
||
max_turns: 50,
|
||
compact_config: Some(CompactConfig::default()),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 对话记忆 —— 按 session 管理多轮对话消息历史。
|
||
///
|
||
/// 内部维护 `Vec<Message>`（Phase 2 IR 类型）作为热缓存，
|
||
/// `MemoryStore` 用作冷持久化层。
|
||
///
|
||
/// ponytail: Phase 2 切换消息存储从 `OpenaiChatMessage` 到 `Message`。
|
||
/// `Message` 已实现 `Serialize` / `Deserialize`（Phase 0 FIX-B 引入），
|
||
/// 序列化格式采用 `#[serde(tag = "type", rename_all = "snake_case")]`，
|
||
/// 例如：
|
||
/// ```json
|
||
/// {"type": "user", "content": [{"type": "text", "text": "hi"}]}
|
||
/// {"type": "tool_result", "tool_call_id": "c1", "content": [...], "is_error": false}
|
||
/// ```
|
||
pub struct ConversationMemory {
|
||
store: Arc<dyn MemoryStore>,
|
||
session_id: String,
|
||
config: ConversationMemoryConfig,
|
||
/// 热缓存：消息列表，供 `llm::compact` 直接操作。
|
||
messages: Vec<Message>,
|
||
/// 与 `messages` 一一对应的存储 ID。
|
||
message_ids: Vec<String>,
|
||
/// 压缩断路器状态。
|
||
compact_state: CompactState,
|
||
}
|
||
|
||
impl ConversationMemory {
|
||
pub fn new(
|
||
store: Arc<dyn MemoryStore>,
|
||
session_id: impl Into<String>,
|
||
config: ConversationMemoryConfig,
|
||
) -> Self {
|
||
Self {
|
||
store,
|
||
session_id: session_id.into(),
|
||
config,
|
||
messages: Vec::new(),
|
||
message_ids: Vec::new(),
|
||
compact_state: CompactState::new(),
|
||
}
|
||
}
|
||
|
||
pub fn session_id(&self) -> &str {
|
||
&self.session_id
|
||
}
|
||
|
||
pub fn config(&self) -> &ConversationMemoryConfig {
|
||
&self.config
|
||
}
|
||
|
||
pub async fn load(&mut self) -> Result<(), MemoryError> {
|
||
let filter = crate::memory::types::MemoryFilter {
|
||
prefix: Some(self.session_prefix()),
|
||
..Default::default()
|
||
};
|
||
let items = self.store.list(&filter).await?;
|
||
let mut pairs: Vec<(String, Message, OffsetDateTime)> = Vec::with_capacity(items.len());
|
||
for item in items {
|
||
match serde_json::from_str::<Message>(&item.content) {
|
||
Ok(msg) => pairs.push((item.id, msg, item.created_at)),
|
||
Err(e) => {
|
||
return Err(MemoryError::Serialization(format!(
|
||
"load message {} failed: {e}",
|
||
item.id
|
||
)));
|
||
}
|
||
}
|
||
}
|
||
pairs.sort_by_key(|p| p.2);
|
||
self.message_ids = pairs.iter().map(|p| p.0.clone()).collect();
|
||
self.messages = pairs.into_iter().map(|p| p.1).collect();
|
||
Ok(())
|
||
}
|
||
|
||
pub async fn add_message(&mut self, msg: Message) -> Result<(), MemoryError> {
|
||
let now = OffsetDateTime::now_utc();
|
||
let index = self.messages.len();
|
||
let id = self.make_message_id(index, &now);
|
||
|
||
// 写入热缓存
|
||
self.messages.push(msg);
|
||
self.message_ids.push(id.clone());
|
||
|
||
// ponytail: 通过 `Message` 的 Serialize 派生实现持久化
|
||
let item = MemoryItem {
|
||
id: id.clone(),
|
||
content: serde_json::to_string(self.messages.last().unwrap())
|
||
.map_err(|e| MemoryError::Serialization(e.to_string()))?,
|
||
metadata: serde_json::json!({ "session_id": &self.session_id, "index": index }),
|
||
created_at: now,
|
||
};
|
||
self.store.save(item).await?;
|
||
|
||
self.maybe_evict_and_compact().await;
|
||
Ok(())
|
||
}
|
||
|
||
pub fn get_history(&self) -> &[Message] {
|
||
&self.messages
|
||
}
|
||
|
||
pub async fn clear(&mut self) -> Result<(), MemoryError> {
|
||
let to_delete = std::mem::take(&mut self.message_ids);
|
||
self.messages.clear();
|
||
self.compact_state = CompactState::new();
|
||
for id in to_delete {
|
||
self.store.delete(&id).await?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub fn len(&self) -> usize {
|
||
self.messages.len()
|
||
}
|
||
|
||
pub fn is_empty(&self) -> bool {
|
||
self.messages.is_empty()
|
||
}
|
||
|
||
fn session_prefix(&self) -> String {
|
||
format!("conv:{}:", self.session_id)
|
||
}
|
||
|
||
fn make_message_id(&self, index: usize, now: &OffsetDateTime) -> String {
|
||
format!(
|
||
"{}{:010}_{}",
|
||
self.session_prefix(),
|
||
index,
|
||
now.unix_timestamp_nanos()
|
||
)
|
||
}
|
||
|
||
async fn maybe_evict_and_compact(&mut self) {
|
||
if self.config.strategy == MemoryStrategy::SlidingWindow {
|
||
while self.messages.len() > self.config.max_turns {
|
||
if let Some(removed_id) = self.message_ids.first().cloned() {
|
||
let _ = self.store.delete(&removed_id).await;
|
||
}
|
||
self.messages.remove(0);
|
||
self.message_ids.remove(0);
|
||
}
|
||
}
|
||
|
||
if let Some(ref compact_config) = self.config.compact_config
|
||
&& should_compact(&self.messages, compact_config, &self.compact_state)
|
||
{
|
||
let keep_recent = compact_config.keep_recent;
|
||
let freed = microcompact(&mut self.messages, keep_recent);
|
||
if freed > 0 {
|
||
self.compact_state.record_success();
|
||
} else {
|
||
let _ = self.compact_state.record_failure();
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use crate::memory::InMemoryStore;
|
||
use crate::memory::MemoryStore;
|
||
|
||
#[tokio::test]
|
||
async fn add_and_get_history() {
|
||
let store = Arc::new(InMemoryStore::new()) as Arc<dyn MemoryStore>;
|
||
let mut conv =
|
||
ConversationMemory::new(store, "session1", ConversationMemoryConfig::default());
|
||
conv.add_message(Message::user_text("hello")).await.unwrap();
|
||
conv.add_message(Message::user_text("world")).await.unwrap();
|
||
assert_eq!(conv.len(), 2);
|
||
assert_eq!(conv.get_history().len(), 2);
|
||
}
|
||
|
||
/// 验证 Message → JSON → Message 往返（包含 ToolResult 等完整信息）。
|
||
#[tokio::test]
|
||
async fn json_roundtrip_preserves_tool_result() {
|
||
let store = Arc::new(InMemoryStore::new()) as Arc<dyn MemoryStore>;
|
||
let mut conv = ConversationMemory::new(store, "s1", ConversationMemoryConfig::default());
|
||
conv.add_message(Message::tool_result("call_1", "ok", false))
|
||
.await
|
||
.unwrap();
|
||
conv.add_message(Message::assistant("done")).await.unwrap();
|
||
|
||
let original = conv.get_history().to_vec();
|
||
assert_eq!(original.len(), 2);
|
||
|
||
// 各变体可序列化 + 反序列化
|
||
for msg in &original {
|
||
let json = serde_json::to_string(msg).unwrap();
|
||
let decoded: Message = serde_json::from_str(&json).unwrap();
|
||
assert_eq!(format!("{:?}", decoded), format!("{:?}", msg));
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn sliding_window_evicts_oldest() {
|
||
let store = Arc::new(InMemoryStore::new()) as Arc<dyn MemoryStore>;
|
||
let config = ConversationMemoryConfig {
|
||
strategy: MemoryStrategy::SlidingWindow,
|
||
max_turns: 3,
|
||
compact_config: None,
|
||
};
|
||
let mut conv = ConversationMemory::new(store, "s1", config);
|
||
for i in 0..5 {
|
||
conv.add_message(Message::user_text(format!("msg-{i}")))
|
||
.await
|
||
.unwrap();
|
||
}
|
||
assert_eq!(conv.len(), 3);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn full_strategy_no_evict() {
|
||
let store = Arc::new(InMemoryStore::new()) as Arc<dyn MemoryStore>;
|
||
let config = ConversationMemoryConfig {
|
||
strategy: MemoryStrategy::Full,
|
||
max_turns: 3,
|
||
compact_config: None,
|
||
};
|
||
let mut conv = ConversationMemory::new(store, "s1", config);
|
||
for i in 0..5 {
|
||
conv.add_message(Message::user_text(format!("msg-{i}")))
|
||
.await
|
||
.unwrap();
|
||
}
|
||
assert_eq!(conv.len(), 5);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn clear_empties_messages() {
|
||
let store = Arc::new(InMemoryStore::new()) as Arc<dyn MemoryStore>;
|
||
let mut conv =
|
||
ConversationMemory::new(store.clone(), "s1", ConversationMemoryConfig::default());
|
||
conv.add_message(Message::user_text("hello")).await.unwrap();
|
||
assert!(!conv.is_empty());
|
||
conv.clear().await.unwrap();
|
||
assert!(conv.is_empty());
|
||
}
|
||
}
|