lc_memory/
with_history.rs

1// lc-memory/src/with_history.rs
2//! LCEL 组合用的"带记忆的 LLM 封装"。
3//!
4//! 把"读记忆 → 拼用户输入 → 调 LLM → 写回记忆"这一组合封装成单个
5//! `Runnable<String, LLMResult>`,让"LLM + 记忆"可以直接进入 LCEL 管道
6//! (pipe 成链、batch、stream 等),不用在业务代码里手写记忆胶水。
7//!
8//! # 两种记忆来源
9//!
10//! - `new(llm, memory)` —— 注入一个具体记忆对象(单会话,原有行为不变)。
11//! - `with_session_history(llm, factory)` —— 注入 **session 回调**(对齐 Python
12//!   `RunnableWithMessageHistory(llm, get_session_history)`):每次调用从
13//!   `config.configurable["session_id"]` 取槽,同一 session 共享历史,不同
14//!   session 互不串扰;缺失 `session_id` 返回 `LcelError::Chain`。
15//!
16//! # 语义
17//!
18//! `invoke(user_input)` 依次执行:
19//! 1. 按模式选定记忆(Shared 直接取;Sessions 按 session_id 取/建槽);
20//! 2. 从记忆读取历史,转成消息(`memory_variables_to_messages`);
21//! 3. 把用户输入作为 Human 消息追加到末尾;
22//! 4. 交给 `llm.chat`(可选 `RunnableConfig` 透传);
23//! 5. 把「用户输入 / 模型回答」写回记忆;
24//! 6. 返回完整 `LLMResult`。
25//!
26//! LLM 错误通过 `L::Error: Into<LcelError>` 进入管道错误;记忆读写错误
27//! 收敛为 `LcelError::Chain`。
28//!
29//! # 泛型
30//!
31//! `L` 是任意实现 `BaseChatModel` 的模型(原生 Provider / `LLMClient` 均可),
32//! 只要其错误类型能转进 `LcelError`(`LLMClient` 天然满足;原生 Provider 见
33//! lc-providers 的 `From<...> for LcelError`)。记忆以 trait 对象持有,任意
34//! `BaseMemory`(Buffer / Window / Summary / SummaryBuffer 等)都可用。
35
36use crate::base::{memory_variables_to_messages, BaseMemory};
37use crate::buffer::ConversationBufferMemory;
38use async_trait::async_trait;
39use lc_core::language_models::{BaseChatModel, LLMResult};
40use lc_core::runnables::{LcelError, Runnable, RunnableConfig};
41use lc_schema::Message;
42use std::collections::{HashMap, VecDeque};
43use std::sync::Arc;
44use tokio::sync::Mutex;
45
46/// 记忆句柄:任意 `BaseMemory` 的共享可变引用。
47pub type SharedMemory = Arc<Mutex<Box<dyn BaseMemory>>>;
48
49/// Session 缓存默认上限:防异常/恶意 session_id 无限增长占满内存(M2a)。
50const DEFAULT_MAX_SESSIONS: usize = 100;
51
52/// 带记忆的 LLM 封装,作为单个 Runnable 参与 LCEL 组合。
53pub struct RunnableWithMessageHistory<L> {
54    llm: Arc<L>,
55    mode: HistoryMode,
56}
57
58/// 记忆来源:共享单槽(SHARED)或按 session 分槽(SESSIONS)。
59enum HistoryMode {
60    /// 构造时注入的单一记忆对象,所有调用共享。
61    Shared(SharedMemory),
62    /// 按 `configurable.session_id` 分槽:factory 建新槽,缓存复用已建槽。
63    Sessions {
64        factory: Arc<dyn Fn(&str) -> SharedMemory + Send + Sync>,
65        /// 槽缓存:同一 session 复用同一份记忆(其锁同时串行化同槽并发 invoke)。
66        cache: Mutex<SessionCache>,
67        /// 缓存上限:超过即淘汰最旧 session,防 session_id 无限增长的内存 DoS(M2a)。
68        max_sessions: usize,
69        /// 仅用于 `memory()` 访问器的占位记忆(Sessions 模式下实际槽不唯一)。
70        default: SharedMemory,
71    },
72}
73
74/// Session 槽缓存:`slots` 按 session_id 取槽,`order` 记录插入顺序供上限淘汰。
75struct SessionCache {
76    slots: HashMap<String, SharedMemory>,
77    order: VecDeque<String>,
78}
79
80impl<L> RunnableWithMessageHistory<L> {
81    /// 用 LLM + 单个记忆对象构造封装(所有调用共享这一份记忆)。
82    pub fn new(llm: L, memory: impl BaseMemory + 'static) -> Self {
83        Self {
84            llm: Arc::new(llm),
85            mode: HistoryMode::Shared(Arc::new(Mutex::new(Box::new(memory)))),
86        }
87    }
88
89    /// 用 LLM + session 回调构造封装(对齐 Python
90    /// `RunnableWithMessageHistory(llm, get_session_history)`)。
91    ///
92    /// `factory(session_id)` 为一个 session 槽建出记忆对象;每次调用按
93    /// `config.configurable["session_id"]` 选槽,同一 session 复用已建槽。
94    ///
95    /// # Example
96    ///
97    /// ```rust,ignore
98    /// let pipe = RunnableWithMessageHistory::with_session_history(llm, |session_id| {
99    ///     Arc::new(Mutex::new(Box::new(
100    ///         ConversationBufferMemory::new().with_return_messages(true),
101    ///     ) as Box<dyn BaseMemory>))
102    /// });
103    /// let cfg = RunnableConfig::new().with_configurable("session_id", json!("s1"));
104    /// pipe.invoke("我叫小明".into(), Some(cfg)).await?;
105    /// ```
106    pub fn with_session_history<F>(llm: L, factory: F) -> Self
107    where
108        F: Fn(&str) -> SharedMemory + Send + Sync + 'static,
109    {
110        Self {
111            llm: Arc::new(llm),
112            mode: HistoryMode::Sessions {
113                factory: Arc::new(factory),
114                cache: Mutex::new(SessionCache {
115                    slots: HashMap::new(),
116                    order: VecDeque::new(),
117                }),
118                max_sessions: DEFAULT_MAX_SESSIONS,
119                default: Arc::new(Mutex::new(Box::new(
120                    ConversationBufferMemory::new(),
121                ))),
122            },
123        }
124    }
125
126    /// 设置 session 缓存上限(Sessions 模式)。超过上限后,新 session 会淘汰最旧
127    /// 的槽,防 session_id 无限增长的内存 DoS(M2a)。默认 [`DEFAULT_MAX_SESSIONS`]。
128    pub fn with_max_sessions(mut self, max: usize) -> Self {
129        if let HistoryMode::Sessions { max_sessions, .. } = &mut self.mode {
130            *max_sessions = max.max(1);
131        }
132        self
133    }
134
135    /// 暴露内部记忆句柄,便于读取已保存的历史(调试、展示、验证写回等)。
136    ///
137    /// Sessions 模式下槽不唯一,返回的是占位记忆(不会参与管道读写);
138    /// 要检查真实历史请从管道内部或业务侧记忆对象读取。
139    pub fn memory(&self) -> SharedMemory {
140        match &self.mode {
141            HistoryMode::Shared(m) => m.clone(),
142            HistoryMode::Sessions { default, .. } => default.clone(),
143        }
144    }
145
146    /// 按模式选定本次调用要用的记忆槽。
147    async fn select_memory(
148        &self,
149        config: &Option<RunnableConfig>,
150    ) -> Result<SharedMemory, LcelError> {
151        match &self.mode {
152            HistoryMode::Shared(m) => Ok(m.clone()),
153            HistoryMode::Sessions {
154                factory,
155                cache,
156                max_sessions,
157                ..
158            } => {
159                let session_id = config
160                    .as_ref()
161                    .and_then(|c| c.configurable_value("session_id"))
162                    .and_then(|v| v.as_str())
163                    .ok_or_else(|| {
164                        LcelError::Chain(
165                            "RunnableWithMessageHistory(session mode) 缺少 configurable.session_id"
166                                .to_string(),
167                        )
168                    })?;
169                let mut cache = cache.lock().await;
170                if let Some(memory) = cache.slots.get(session_id) {
171                    return Ok(memory.clone());
172                }
173                // M2a: 缓存有上限,先淘汰最旧的 session,再建新槽。
174                if cache.slots.len() >= *max_sessions {
175                    if let Some(oldest) = cache.order.pop_front() {
176                        cache.slots.remove(&oldest);
177                    }
178                }
179                let memory = factory(session_id);
180                cache
181                    .slots
182                    .insert(session_id.to_string(), memory.clone());
183                cache.order.push_back(session_id.to_string());
184                Ok(memory)
185            }
186        }
187    }
188}
189
190#[async_trait]
191impl<L> Runnable<String, LLMResult> for RunnableWithMessageHistory<L>
192where
193    L: BaseChatModel + 'static,
194    L::Error: Into<LcelError>,
195{
196    type Error = LcelError;
197
198    async fn invoke(
199        &self,
200        input: String,
201        config: Option<RunnableConfig>,
202    ) -> Result<LLMResult, LcelError> {
203        // 0. 选定记忆槽(Sessions 模式读 configurable.session_id)
204        let memory = self.select_memory(&config).await?;
205
206        // M2b: 整个「读记忆 → 调 LLM → 写回」持锁执行,串行化同一记忆槽的并发
207        // invoke。旧实现读后即放锁,两个并发 invoke 都读到旧历史、写回互相覆盖
208        // 丢历史。代价是同槽调用串行,但带记忆的对话本就应串行。
209        let mut memory = memory.lock().await;
210
211        // 1. 读记忆 → 转消息
212        let mut messages = {
213            let vars = memory
214                .load_memory_variables(&HashMap::new())
215                .await
216                .map_err(|e| LcelError::Chain(format!("load memory: {e}")))?;
217            memory_variables_to_messages(&vars)
218        };
219
220        // 2. 拼上用户输入
221        messages.push(Message::human(&input));
222
223        // 3. 调 LLM(持锁等待:同槽串行,避免并发丢历史)
224        let result = self
225            .llm
226            .chat(messages, config)
227            .await
228            .map_err(Into::into)?;
229
230        // 4. 写回记忆:失败不丢弃模型答案(否则调用方拿 Err 重试会重复调 LLM),
231        //    记 warn 暴露记忆层降级
232        let inputs = HashMap::from([("input".to_string(), input)]);
233        let outputs = HashMap::from([("output".to_string(), result.content.clone())]);
234        if let Err(e) = memory.save_context(&inputs, &outputs).await {
235            log::warn!("记忆写回失败(模型答案仍照常返回): {e}");
236        }
237
238        Ok(result)
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245    use futures_util::Stream;
246    use lc_core::language_models::{BaseChatModel, BaseLanguageModel};
247    use lc_core::runnables::RunnableConfig;
248    use lc_schema::MessageType;
249    use serde_json::json;
250    use std::pin::Pin;
251    use std::sync::Mutex as StdMutex;
252
253    /// 测试 LLM:记录每次收到的消息,并把最后一条用户消息包成回答。
254    struct TestChatModel {
255        seen: Arc<StdMutex<Vec<Vec<Message>>>>,
256    }
257
258    #[async_trait]
259    impl Runnable<Vec<Message>, LLMResult> for TestChatModel {
260        type Error = LcelError;
261
262        async fn invoke(
263            &self,
264            input: Vec<Message>,
265            _config: Option<RunnableConfig>,
266        ) -> Result<LLMResult, LcelError> {
267            self.seen.lock().unwrap().push(input.clone());
268            let last = input
269                .last()
270                .map(|m| m.content.clone())
271                .unwrap_or_default();
272            Ok(LLMResult {
273                content: format!("reply to: {last}"),
274                ..Default::default()
275            })
276        }
277    }
278
279    #[async_trait]
280    impl BaseLanguageModel<Vec<Message>, LLMResult> for TestChatModel {
281        fn model_name(&self) -> &str {
282            "test-llm"
283        }
284
285        fn get_num_tokens(&self, text: &str) -> usize {
286            text.len()
287        }
288
289        fn with_temperature(self, _temp: f32) -> Self
290        where
291            Self: Sized,
292        {
293            self
294        }
295
296        fn with_max_tokens(self, _max: usize) -> Self
297        where
298            Self: Sized,
299        {
300            self
301        }
302    }
303
304    #[async_trait]
305    impl BaseChatModel for TestChatModel {
306        async fn chat(
307            &self,
308            messages: Vec<Message>,
309            config: Option<RunnableConfig>,
310        ) -> Result<LLMResult, LcelError> {
311            self.invoke(messages, config).await
312        }
313
314        async fn stream_chat(
315            &self,
316            _messages: Vec<Message>,
317            _config: Option<RunnableConfig>,
318        ) -> Result<
319            Pin<Box<dyn Stream<Item = Result<String, LcelError>> + Send>>,
320            LcelError,
321        > {
322            unimplemented!("stream_chat not needed for tests")
323        }
324    }
325
326    /// session 回调:每次建一个空的 Buffer 记忆(return_messages = true)。
327    fn session_factory(
328        _session_id: &str,
329    ) -> SharedMemory {
330        Arc::new(Mutex::new(Box::new(
331            ConversationBufferMemory::new().with_return_messages(true),
332        ) as Box<dyn BaseMemory>))
333    }
334
335    /// 读记忆 → LLM → 写回:第二轮调用时,LLM 应看到第一轮的完整对话。
336    #[tokio::test]
337    async fn reads_memory_writes_back_round_trip() {
338        // Arrange
339        let seen = Arc::new(StdMutex::new(Vec::new()));
340        let llm = TestChatModel { seen: seen.clone() };
341        // return_messages = true:历史以消息数组返回,方便断言每轮消息构成
342        let memory = ConversationBufferMemory::new().with_return_messages(true);
343        let pipe = RunnableWithMessageHistory::new(llm, memory);
344
345        // Act 第一轮:无历史
346        let r1 = pipe.invoke("我叫什么名字".to_string(), None).await.unwrap();
347        assert_eq!(r1.content, "reply to: 我叫什么名字");
348
349        // Act 第二轮:历史应已写回
350        let r2 = pipe.invoke("再问一次".to_string(), None).await.unwrap();
351        assert_eq!(r2.content, "reply to: 再问一次");
352
353        // Assert
354        let calls = seen.lock().unwrap();
355        assert_eq!(calls.len(), 2, "应调用模型两次");
356        // 第一轮:只有用户消息
357        assert_eq!(calls[0].len(), 1);
358        assert_eq!(calls[0][0].content, "我叫什么名字");
359        // 第二轮:user + ai + 新 user(记忆已写回)
360        assert_eq!(calls[1].len(), 3);
361        assert_eq!(calls[1][0].content, "我叫什么名字");
362        assert!(matches!(calls[1][0].message_type, MessageType::Human));
363        assert!(matches!(calls[1][1].message_type, MessageType::AI));
364        assert_eq!(calls[1][2].content, "再问一次");
365    }
366
367    /// 记忆写回持久化在封装内:构造新封装、复用同一记忆类型,历史仍在。
368    #[tokio::test]
369    async fn memory_accumulates_across_invocations() {
370        // Arrange
371        let seen = Arc::new(StdMutex::new(Vec::new()));
372        let llm = TestChatModel { seen: seen.clone() };
373        let memory = ConversationBufferMemory::new().with_return_messages(true);
374        let pipe = RunnableWithMessageHistory::new(llm, memory);
375
376        // Act 三轮连续调用
377        for turn in ["你好", "你在吗", "再见"] {
378            pipe.invoke(turn.to_string(), None).await.unwrap();
379        }
380
381        // Assert 第三轮应看到前两轮完整四段对话
382        let calls = seen.lock().unwrap();
383        assert_eq!(calls.len(), 3);
384        assert_eq!(calls[2].len(), 5); // 2 轮 * 2 段 + 当前用户消息
385        assert_eq!(calls[2][4].content, "再见");
386    }
387
388    /// session 模式:同一 session_id 两轮调用共享历史。
389    #[tokio::test]
390    async fn session_history_same_session_shares_memory() {
391        // Arrange
392        let seen = Arc::new(StdMutex::new(Vec::new()));
393        let llm = TestChatModel { seen: seen.clone() };
394        let pipe = RunnableWithMessageHistory::with_session_history(llm, session_factory);
395        let cfg = RunnableConfig::new().with_configurable("session_id", json!("s1"));
396
397        // Act 同一 session 两轮
398        let r1 = pipe.invoke("我叫什么名字".to_string(), Some(cfg.clone())).await.unwrap();
399        assert_eq!(r1.content, "reply to: 我叫什么名字");
400        let r2 = pipe.invoke("再问一次".to_string(), Some(cfg)).await.unwrap();
401        assert_eq!(r2.content, "reply to: 再问一次");
402
403        // Assert 第二轮看到第一轮完整对话(user + ai + user)
404        let calls = seen.lock().unwrap();
405        assert_eq!(calls.len(), 2);
406        assert_eq!(calls[1].len(), 3);
407        assert_eq!(calls[1][0].content, "我叫什么名字");
408        assert!(matches!(calls[1][1].message_type, MessageType::AI));
409    }
410
411    /// session 模式:不同 session_id 互不串扰。
412    #[tokio::test]
413    async fn session_history_different_sessions_isolated() {
414        // Arrange
415        let seen = Arc::new(StdMutex::new(Vec::new()));
416        let llm = TestChatModel { seen: seen.clone() };
417        let pipe = RunnableWithMessageHistory::with_session_history(llm, session_factory);
418        let cfg_s1 = RunnableConfig::new().with_configurable("session_id", json!("s1"));
419        let cfg_s2 = RunnableConfig::new().with_configurable("session_id", json!("s2"));
420
421        // Act s1 两轮 + s2 一轮
422        pipe.invoke("我是 s1".to_string(), Some(cfg_s1.clone())).await.unwrap();
423        pipe.invoke("还在 s1".to_string(), Some(cfg_s1)).await.unwrap();
424        pipe.invoke("我是 s2".to_string(), Some(cfg_s2)).await.unwrap();
425
426        // Assert s2 第一轮无历史(1 条),s1 第二轮有历史(3 条)
427        let calls = seen.lock().unwrap();
428        assert_eq!(calls.len(), 3);
429        assert_eq!(calls[1].len(), 3, "s1 第二轮应看到历史");
430        assert_eq!(calls[2].len(), 1, "s2 第一轮应无历史");
431    }
432
433    /// session 模式:缺失 configurable.session_id → LcelError::Chain。
434    #[tokio::test]
435    async fn session_history_missing_session_id_errors() {
436        // Arrange
437        let seen = Arc::new(StdMutex::new(Vec::new()));
438        let llm = TestChatModel { seen };
439        let pipe = RunnableWithMessageHistory::with_session_history(llm, session_factory);
440
441        // Act 无 config / 无 session_id
442        let err = pipe.invoke("你好".to_string(), None).await.unwrap_err();
443        assert!(matches!(err, LcelError::Chain(_)));
444
445        let cfg_no_sid = RunnableConfig::new();
446        let err = pipe
447            .invoke("你好".to_string(), Some(cfg_no_sid))
448            .await
449            .unwrap_err();
450        assert!(matches!(err, LcelError::Chain(_)));
451    }
452
453    /// 可阻塞的测试 LLM:第一次 chat 通知 `entered` 并阻塞在 `release`,用于在
454    /// 并发测试中制造「已持锁、模型调用中」的确定性窗口(M2b)。
455    struct BlockingChatModel {
456        seen: Arc<StdMutex<Vec<Vec<Message>>>>,
457        entered: Arc<tokio::sync::Notify>,
458        release: Arc<tokio::sync::Notify>,
459        blocked: Arc<StdMutex<bool>>,
460    }
461
462    #[async_trait]
463    impl Runnable<Vec<Message>, LLMResult> for BlockingChatModel {
464        type Error = LcelError;
465
466        async fn invoke(
467            &self,
468            input: Vec<Message>,
469            _config: Option<RunnableConfig>,
470        ) -> Result<LLMResult, LcelError> {
471            self.seen.lock().unwrap().push(input.clone());
472            let should_block = {
473                let mut blocked = self.blocked.lock().unwrap();
474                if !*blocked {
475                    *blocked = true;
476                    true
477                } else {
478                    false
479                }
480            };
481            self.entered.notify_one();
482            if should_block {
483                self.release.notified().await;
484            }
485            let last = input
486                .last()
487                .map(|m| m.content.clone())
488                .unwrap_or_default();
489            Ok(LLMResult {
490                content: format!("reply to: {last}"),
491                ..Default::default()
492            })
493        }
494    }
495
496    #[async_trait]
497    impl BaseLanguageModel<Vec<Message>, LLMResult> for BlockingChatModel {
498        fn model_name(&self) -> &str {
499            "blocking-test-llm"
500        }
501
502        fn get_num_tokens(&self, text: &str) -> usize {
503            text.len()
504        }
505
506        fn with_temperature(self, _temp: f32) -> Self
507        where
508            Self: Sized,
509        {
510            self
511        }
512
513        fn with_max_tokens(self, _max: usize) -> Self
514        where
515            Self: Sized,
516        {
517            self
518        }
519    }
520
521    #[async_trait]
522    impl BaseChatModel for BlockingChatModel {
523        async fn chat(
524            &self,
525            messages: Vec<Message>,
526            config: Option<RunnableConfig>,
527        ) -> Result<LLMResult, LcelError> {
528            self.invoke(messages, config).await
529        }
530
531        async fn stream_chat(
532            &self,
533            _messages: Vec<Message>,
534            _config: Option<RunnableConfig>,
535        ) -> Result<
536            Pin<Box<dyn Stream<Item = Result<String, LcelError>> + Send>>,
537            LcelError,
538        > {
539            unimplemented!("stream_chat not needed for tests")
540        }
541    }
542
543    /// session 缓存有上限:超过后淘汰最旧 session,防 session_id 无限增长的内存 DoS(M2a)。
544    #[tokio::test]
545    async fn session_cache_evicts_oldest_when_over_capacity() {
546        // Arrange
547        let seen = Arc::new(StdMutex::new(Vec::new()));
548        let llm = TestChatModel { seen: seen.clone() };
549        let pipe = RunnableWithMessageHistory::with_session_history(llm, session_factory)
550            .with_max_sessions(2);
551
552        let cfg_s1 = RunnableConfig::new().with_configurable("session_id", json!("s1"));
553        let cfg_s2 = RunnableConfig::new().with_configurable("session_id", json!("s2"));
554        let cfg_s3 = RunnableConfig::new().with_configurable("session_id", json!("s3"));
555
556        // Act s1、s2 占满缓存;s3 触发淘汰最旧的 s1。
557        pipe.invoke("s1-turn1".to_string(), Some(cfg_s1.clone()))
558            .await
559            .unwrap();
560        pipe.invoke("s2-turn1".to_string(), Some(cfg_s2))
561            .await
562            .unwrap();
563        pipe.invoke("s3-turn1".to_string(), Some(cfg_s3))
564            .await
565            .unwrap();
566        // s1 已被淘汰 → 重入 s1 是全新会话。
567        pipe.invoke("s1-turn2".to_string(), Some(cfg_s1))
568            .await
569            .unwrap();
570
571        // Assert
572        let calls = seen.lock().unwrap();
573        assert_eq!(calls.len(), 4);
574        assert_eq!(
575            calls[3].len(),
576            1,
577            "M2a: s1 槽被淘汰后重入应为全新会话(无历史)"
578        );
579    }
580
581    /// 同一记忆槽的并发 invoke 必须串行化:读→LLM→写整段持锁,避免并发都读到
582    /// 旧历史、写回互相覆盖丢上下文(M2b)。
583    #[tokio::test]
584    async fn concurrent_same_session_invokes_do_not_lose_history() {
585        // Arrange
586        let seen = Arc::new(StdMutex::new(Vec::new()));
587        let entered = Arc::new(tokio::sync::Notify::new());
588        let release = Arc::new(tokio::sync::Notify::new());
589        let blocked = Arc::new(StdMutex::new(false));
590        let llm = BlockingChatModel {
591            seen: seen.clone(),
592            entered: entered.clone(),
593            release: release.clone(),
594            blocked,
595        };
596        let pipe = Arc::new(RunnableWithMessageHistory::new(
597            llm,
598            ConversationBufferMemory::new().with_return_messages(true),
599        ));
600
601        // Act 第一轮 invoke:进入模型调用后阻塞,期间必须持有记忆锁。
602        let p1 = pipe.clone();
603        let h1 = tokio::spawn(async move { p1.invoke("第一轮".to_string(), None).await });
604        entered.notified().await;
605
606        // 第二轮 invoke:若读→LLM→写没有整段持锁,此刻会读到空历史(丢上下文)。
607        let p2 = pipe.clone();
608        let h2 = tokio::spawn(async move { p2.invoke("第二轮".to_string(), None).await });
609
610        // 放行第一轮:写回后才释放锁,第二轮才能读到第一轮完整对话。
611        release.notify_one();
612        let r1 = h1.await.unwrap().unwrap();
613        let r2 = h2.await.unwrap().unwrap();
614        assert_eq!(r1.content, "reply to: 第一轮");
615        assert_eq!(r2.content, "reply to: 第二轮");
616
617        // Assert
618        let calls = seen.lock().unwrap();
619        assert_eq!(calls.len(), 2);
620        assert_eq!(
621            calls[1].len(),
622            3,
623            "M2b: 第二轮应看到第一轮完整对话(user+ai+user),而非空历史"
624        );
625        assert_eq!(calls[1][0].content, "第一轮");
626        assert_eq!(calls[1][2].content, "第二轮");
627    }
628}
lc_memory/with_history.rs

lc_memory/
with_history.rs
