腾讯云数据库 AI 服务异步慢召回（deepsearch）

一、慢召回（DeepSearch）

·检索精准度高：深度挖掘相关记忆内容。

·信息覆盖面广：全量扫描记忆数据。

·具备推理能力：支持多跳推理分析。

·异步处理机制：避免用户界面阻塞。

二、适用场景

·复杂推理任务。

·综合分析需求。

·非即时反馈的闲聊场景。

三、异步交互设计

采用"双线程 + 预加载"策略（参考 async_retrieval.py 实现），确保流畅的用户体验：

·主线程 (Chat Loop)

功能	描述	实现方式
用户交互处理	负责处理用户输入和 LLM 实时回复	实时响应机制
记忆注入	将上一轮 DeepSearch 结果作为 Context 注入当前轮对话	预加载记忆数据
体验保障	确保用户界面流畅，无需等待记忆搜索完成	非阻塞交互设计

·后台线程 (Memory Worker)

功能	描述	实现方式
静默搜索	后台运行 DeepSearch 任务	独立线程处理
深度挖掘	根据用户问题深入挖掘相关记忆	多跳推理算法
结果缓存	将挖掘结果存储到 pending_memory 供下一轮使用	预加载机制

四、AsyncMemoryManager 类：管理后台线程和共享状态锁

class AsyncMemoryManager: def __init__(self, actor_id: str): # … 初始化 … self.pending_memory: Optional[str] = None self.is_searching = False self.lock = threading.Lock() def start_deepsearch(self, query: str): # 启动后台线程进行 DeepSearch thread = threading.Thread(target=self._deepsearch_worker, args=(query,), daemon=True) thread.start() def _deepsearch_worker(self, query: str): try: # 执行耗时的 deepsearch result = self.client.deepsearch( query=query, actor_id=self.actor_id, summary_by_model=True ) formatted = self._format_deepsearch_result(result) with self.lock: # 将结果存入缓存，供主线程下一轮读取 if formatted: self.pending_memory = formatted finally: self.is_searching = False

五、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: Python 3.11.1

六、主交互循环：实现无阻塞聊天体验

def main(): user_id = "user_a" memory_manager = AsyncMemoryManager(actor_id=user_id) while True: user_input = input(f"\\n请输入你的问题: ").strip() # 1. 获取上一轮后台挖掘到的记忆 (如果有) pending_memory = memory_manager.get_and_clear_pending_memory() if pending_memory: print(f"\\n[提示] 使用上轮 DeepSearch 召回的记忆进行回复…") # 2. 立即回复用户 (带上记忆Context) # 此时 DeepSearch 还在后台可能没跑完，但不影响这次回复（除非有 pending） chat_with_crewai_streaming( user_message=user_input, memory_context=pending_memory ) # 3. 触发新一轮的后台挖掘，为后续聊天做准备 memory_manager.start_deepsearch(user_input)

七、Running Environment