fix(api): use batch memory ingest to prevent seq ordering race condition - #1998
Open
lanceyq wants to merge 3 commits into
Open
fix(api): use batch memory ingest to prevent seq ordering race condition#1998lanceyq wants to merge 3 commits into
lanceyq wants to merge 3 commits into
Conversation
Contributor
审阅者指南引入了一条新的代理消息批量写入路径，用于在单个事务中写入记忆消息，并在一次滑动窗口遍历中完成分发，从而确保用户/助手消息对的序列号严格有序分配，并围绕新 API 提供“发出即忘”（fire-and-forget）的后台分发辅助工具。 代理消息批量写入与分发的序列图sequenceDiagram
participant AppChatService
participant ConversationService
participant fire_background_memory_task
participant MemoryService
participant dispatcher as ingest_agent_messages
participant Repo as MemoryMessageRepository
participant FastWrite as safe_push_fast_write
participant Sliding as check_sliding_window_and_dispatch
AppChatService->>ConversationService: dispatch_memory_batch(messages, conversation)
ConversationService->>fire_background_memory_task: fire_background_memory_task(coro)
fire_background_memory_task-->>ConversationService: Task
ConversationService->>MemoryService: ingest_agent_messages(conversation_id, messages, app_id, config_id, workspace_id, end_user_id)
MemoryService->>dispatcher: ingest_agent_messages(conversation_id, messages, app_id, config_id, workspace_id, end_user_id, language)
dispatcher->>Repo: write_batch(conversation_id, messages, end_user_id, source=AGENT)
Repo-->>dispatcher: written[message_seq]
dispatcher->>dispatcher: refresh_active_key(conversation_id)
dispatcher->>dispatcher: mark_conversation_pending(conversation_id)
loop for each written message
dispatcher->>FastWrite: safe_push_fast_write(role, should_memorize, app_id, end_user_id, target_message, config_id, workspace_id, conversation_id, message_seq, language, source)
end
dispatcher->>Sliding: check_sliding_window_and_dispatch(conversation_id, config_id, end_user_id, workspace_id, language)
Sliding-->>dispatcher: dispatch_done
dispatcher-->>MemoryService: True
MemoryService-->>ConversationService: True
文件级变更
提示与命令与 Sourcery 交互
自定义你的使用体验访问你的 dashboard 来：
获取帮助Original review guide in EnglishReviewer's GuideIntroduces a new batch ingest path for agent messages to write memory messages in a single transaction and dispatch in one sliding-window pass, ensuring strictly ordered seq assignment for user/assistant pairs and providing a fire-and-forget background dispatch helper around the new API. Sequence diagram for batch agent message ingest and dispatchsequenceDiagram
participant AppChatService
participant ConversationService
participant fire_background_memory_task
participant MemoryService
participant dispatcher as ingest_agent_messages
participant Repo as MemoryMessageRepository
participant FastWrite as safe_push_fast_write
participant Sliding as check_sliding_window_and_dispatch
AppChatService->>ConversationService: dispatch_memory_batch(messages, conversation)
ConversationService->>fire_background_memory_task: fire_background_memory_task(coro)
fire_background_memory_task-->>ConversationService: Task
ConversationService->>MemoryService: ingest_agent_messages(conversation_id, messages, app_id, config_id, workspace_id, end_user_id)
MemoryService->>dispatcher: ingest_agent_messages(conversation_id, messages, app_id, config_id, workspace_id, end_user_id, language)
dispatcher->>Repo: write_batch(conversation_id, messages, end_user_id, source=AGENT)
Repo-->>dispatcher: written[message_seq]
dispatcher->>dispatcher: refresh_active_key(conversation_id)
dispatcher->>dispatcher: mark_conversation_pending(conversation_id)
loop for each written message
dispatcher->>FastWrite: safe_push_fast_write(role, should_memorize, app_id, end_user_id, target_message, config_id, workspace_id, conversation_id, message_seq, language, source)
end
dispatcher->>Sliding: check_sliding_window_and_dispatch(conversation_id, config_id, end_user_id, workspace_id, language)
Sliding-->>dispatcher: dispatch_done
dispatcher-->>MemoryService: True
MemoryService-->>ConversationService: True
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
嗨，我在这里给出了一些高层次的反馈：
- 在
dispatch_memory_batch中，logger.warning(..., exc_info=exc)调用了exc_info时传入的是异常对象而不是布尔值/元组；这样不会记录到期望的 traceback——请改用exc_info=True，或者传入正确的(type, value, traceback)元组。 - 全局的
_BACKGROUND_MEMORY_TASKS集合依赖任务最终完成后被移除；建议考虑针对生命周期极长或卡住的任务做保护（例如使用超时或定期清理），以避免在长时间运行的进程中集合无限增长。 - 当没有事件循环在运行时，
fire_background_memory_task会静默地关闭协程，这可能会隐藏同步环境下的错误用法；与其丢弃这些工作，你或许应该抛出或以某种方式向调用方暴露这个错误。
供 AI 代理使用的提示
请根据这次代码评审中的评论进行修改：
## 总体评论
- 在 `dispatch_memory_batch` 中，`logger.warning(..., exc_info=exc)` 调用了 `exc_info` 时传入的是异常对象而不是布尔值/元组；这样不会记录到期望的 traceback——请改用 `exc_info=True`，或者传入正确的 `(type, value, traceback)` 元组。
- 全局的 `_BACKGROUND_MEMORY_TASKS` 集合依赖任务最终完成后被移除；建议考虑针对生命周期极长或卡住的任务做保护（例如使用超时或定期清理），以避免在长时间运行的进程中集合无限增长。
- 当没有事件循环在运行时，`fire_background_memory_task` 会静默地关闭协程，这可能会隐藏同步环境下的错误用法；与其丢弃这些工作，你或许应该抛出或以某种方式向调用方暴露这个错误。帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've left some high level feedback:
- In
dispatch_memory_batch, thelogger.warning(..., exc_info=exc)call passes the exception object instead of a bool/tuple; this will not log the intended traceback—useexc_info=Trueor a proper(type, value, traceback)tuple. - The global
_BACKGROUND_MEMORY_TASKSset relies on tasks eventually completing to be discarded; consider guarding against very long‑lived or stuck tasks (e.g., with timeouts or periodic cleanup) to avoid unbounded growth in long‑running processes. fire_background_memory_tasksilently closes the coroutine when no event loop is running, which can hide incorrect use from synchronous contexts; you may want to raise or surface this as an error to callers instead of discarding the work.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `dispatch_memory_batch`, the `logger.warning(..., exc_info=exc)` call passes the exception object instead of a bool/tuple; this will not log the intended traceback—use `exc_info=True` or a proper `(type, value, traceback)` tuple.
- The global `_BACKGROUND_MEMORY_TASKS` set relies on tasks eventually completing to be discarded; consider guarding against very long‑lived or stuck tasks (e.g., with timeouts or periodic cleanup) to avoid unbounded growth in long‑running processes.
- `fire_background_memory_task` silently closes the coroutine when no event loop is running, which can hide incorrect use from synchronous contexts; you may want to raise or surface this as an error to callers instead of discarding the work.Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
API Breaking Change ReportComparing against Breaking change approval label:
oasdiffOutputopenapi-diffOutputGate decisionNo breaking changes detected. The check passed without approval override. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
引入对代理消息的批量写入和调度到记忆系统的能力，以确保顺序一致的串行排序，并减少用户与助手消息之间的竞态条件。
New Features:
ConversationService.batch的记忆派发 API，以及一个“即发即弃”（fire-and-forget）助手方法，用于在不阻塞聊天流程的情况下调度后台记忆任务。Bug Fixes:
Enhancements:
MemoryService中封装新的批量记忆写入逻辑，便于上层服务集中使用。Original summary in English
Summary by Sourcery
Introduce batch ingestion and dispatch of agent messages to the memory system to ensure consistent sequential ordering and reduce race conditions between user and assistant messages.
New Features:
Bug Fixes:
Enhancements: