从零构建支持中断恢复的AI Agent：状态机与记忆系统设计实践

1. 项目概述：为什么我们需要一个能“被打断”的AI助手？

最近在折腾一个挺有意思的项目，核心目标就写在标题里了：从零开始搭建一个能理解“人机协作”与“中断恢复”的AI Agent。这听起来可能有点抽象，但如果你用过市面上那些“一问一答”式的聊天机器人，肯定遇到过这样的场景：你正在让它帮你写一份报告，刚列好大纲，老板突然在群里@你，让你立刻处理一个紧急数据。你不得不中断和AI的对话，等忙完回来，要么得从头开始描述你的需求，要么AI已经“失忆”，完全不记得刚才的上下文了。

这就是传统AI交互模式的痛点——它假设对话是线性的、连续的，但真实世界的工作流充满了中断、并行和优先级切换。一个真正有用的AI助手，不应该只是一个更聪明的搜索引擎，而应该像一个得力的工作伙伴，能理解任务的上下文，在你离开后能“暂停”，在你回来时能“续上”，甚至能主动管理多个并行的任务线程。这就是我动手搭建这个AI Agent的初衷：探索如何让AI具备基础的“工作记忆”和“任务状态管理”能力，实现更自然、更高效的人机协作。

这个项目不依赖于任何单一的、庞大的语言模型，而是侧重于架构设计与流程控制。我们会用一些开源的轻量级模型和框架作为核心组件，但更重要的是构建一套让这些组件协同工作的逻辑。最终实现的Agent，将能够接受一个复杂任务（比如“帮我策划一次团队outing”），将其分解为多个步骤（定主题、选地点、做预算、发通知），并允许你在任何步骤介入、修改或暂停，Agent会记住所有状态，确保任务最终被完整、正确地执行。

2. 核心设计思路：将“协作”与“中断”机制化

2.1 打破线性对话：任务即状态机

第一个要摒弃的观念，就是把AI对话看作一连串的Q&A。在我们的设计中，每一个用户发起的顶层任务，都被实例化为一个独立的状态机（State Machine）。这个状态机有明确的生命周期：创建 -> 规划 -> 执行 -> 暂停 -> 恢复 -> 完成/取消。

为什么是状态机？因为它天然适合描述有步骤、可中断的过程。比如“写周报”这个任务，其状态可能包括：等待输入原始数据、生成初稿、等待用户审阅、根据反馈修改、最终定稿。当用户说“等一下，我先发个邮件”，Agent不应该清空“写周报”的所有中间数据，而仅仅是将状态从执行切换到暂停，并持久化保存当前的进度（比如已经生成的初稿内容、收集到的数据点）。

注意：这里的状态持久化是关键。你不能只把状态存在内存里，一旦服务重启就全丢了。最简单的做法是使用一个轻量级数据库（如SQLite）或甚至一个JSON文件来记录每个任务ID对应的状态快照（snapshot）。快照里需要包含：任务目标、已完成的步骤、当前的输出、下一步的指令、以及相关的上下文数据。

2.2 构建智能体的“工作记忆”：上下文管理与会话线程

要让AI记住“刚才说到哪了”，我们需要设计一个分层的上下文管理系统。这不仅仅是把最近的几条对话记录塞给模型那么简单。

1. 会话线程（Conversation Thread）：每个独立的任务就是一个线程。用户和AI围绕这个任务的所有交互，都归属于这个线程。线程ID是唯一的，便于检索和恢复。
2. 短期工作记忆（Short-term Working Memory）：这是模型上下文窗口（比如GPT的16K、32K tokens）直接能“看到”的内容。我们只在这里存放最相关、最精简的信息，例如：当前步骤的指令、上一步的输出、以及关键的决策历史。目的是节省宝贵的Token，并提高模型对当前任务的专注度。
3. 长期记忆（Long-term Memory）：这是存储在外部向量数据库（如ChromaDB, FAISS）里的东西。当对话进行时，我们把重要的中间结论、用户提供的详细资料、生成的文档片段，转换成向量（Embeddings）存起来。当Agent需要“回忆”某个细节时（比如用户问“刚才我们定的预算是多少？”），它可以通过语义搜索从长期记忆中快速检索出相关片段，再注入到短期记忆中供模型使用。

实操心得：很多新手会犯一个错误，就是把整个冗长的对话历史都扔进短期记忆。这会导致模型注意力分散，并且很快耗尽上下文长度。正确的做法是进行摘要（Summarization）。每完成一个任务步骤或经过一定轮次的对话，就用一个小模型（或调用大模型的摘要功能）对当前进展做一次摘要，用这个摘要来更新短期记忆的“头部”，而将原始细节移入长期记忆。这样既能保持连贯性，又能控制上下文长度。

2.3 设计中断与恢复协议：明确的信号与清晰的边界

中断不是错误，而是一种正常的操作指令。因此，我们需要为Agent定义一套清晰的中断与恢复协议。

• 中断信号：用户可以通过特定指令（如“/pause”、“稍等，我先处理点别的”）或简单地开启一个全新主题的对话来发出中断信号。Agent需要能识别这些信号。
• 状态保存点（Checkpoint）：收到中断信号后，Agent应立即触发一个“保存点”操作。这包括：
  • 将当前任务状态机的状态序列化保存。
  • 将短期工作记忆中的关键上下文进行摘要并保存。
  • 记录下导致中断的用户查询（如果有的话），作为恢复时的参考。
• 恢复机制：当用户回来，通过指令（如“/resume [任务ID]”、“继续刚才写周报的事”）或语义匹配（用户问“我们刚才说到哪了？”）要求恢复时，Agent需要：
  • 根据任务ID或语义搜索找到对应的任务上下文。
  • 加载保存的状态和记忆摘要。
  • 生成一条友好的恢复提示，例如：“欢迎回来！我们刚才正在‘撰写Q2项目复盘周报’。我已经完成了背景部分和主要成就的初稿，接下来准备写‘遇到的挑战与改进’。我们从这里继续，好吗？”

这套协议的核心，是让中断和恢复变得可预测、可管理，而不是一次意外的对话崩溃。

3. 技术栈选型与核心模块拆解

搭建这样一个Agent，我们不需要从头造轮子，合理利用现有开源生态是关键。以下是我的技术选型思路，兼顾了能力、复杂度和学习成本。

3.1 大脑核心：轻量级与功能型模型搭配

完全依赖GPT-4这样的大型商用API，成本高且不利于深度定制。我采用混合模式：

1. 规划与调度模型（Planner）：选用Llama 3.1 8B或Qwen 2.5 7B这类中等规模的开源模型。它们的推理能力足够强，可以理解复杂任务，并将其分解成清晰的步骤列表（Step-by-step Plan）。这个模型负责“想”，即任务分解和步骤规划。我们可以使用LM Studio或Ollama在本地运行它，响应快且隐私性好。
2. 执行与对话模型（Executor）：对于每一步的具体执行，比如根据大纲写一段文字、写一段Python代码、或者回答用户的具体问题，可以视情况选择。对质量要求高的步骤，可以调用DeepSeek-V3或GPT-3.5-Turbo的API；对简单、格式化的响应，可以用更小的本地模型（如Phi-3-mini）。这个模型负责“做”，即执行具体指令。
3. 摘要与工具调用模型（Helper）：用于生成对话摘要、判断用户意图（是提问、指令还是中断信号）、以及决定是否需要调用外部工具（如计算器、搜索引擎）。这个角色对能力要求相对较低，可以使用非常轻量的模型（如Gemma 2B），以降低整体延迟和资源消耗。

为什么这么选？将“规划”和“执行”分离，符合人类的思考方式，也使得系统更模块化、更易调试。规划模型一次生成整个计划，执行模型则专注于当前步骤，互不干扰。同时，混合使用本地模型和云端API，在成本、性能和隐私之间取得了平衡。

3.2 记忆系统：向量数据库与结构化存储

1. 向量数据库（长期记忆）：我选择ChromaDB。它轻量、易用，Python集成度极高，非常适合原型和中小型项目。我们将每个任务线程的所有“记忆片段”（用户输入、AI输出、生成的文档块）都通过text-embedding-3-small这类嵌入模型转换成向量，存入ChromaDB的对应集合（Collection）中，集合名可以用任务ID命名。
2. 结构化存储（任务状态）：使用SQLite足矣。我们需要一张tasks表，核心字段包括：
   • task_id(主键)
   • user_id(区分不同用户)
   • goal(任务目标)
   • current_state(如 “planning”, “executing_step_3”, “paused”)
   • plan(JSON格式存储的任务分解步骤)
   • checkpoint(JSON格式存储的进度快照，如上一步输出)
   • created_at,updated_at

3.3 控制中枢：Agent框架与流程编排

虽然可以完全手写控制逻辑，但使用一个成熟的Agent框架能事半功倍。这里我推荐LangChain或LlamaIndex。

• LangChain：优势在于其丰富的“链”（Chain）和“智能体”（Agent）抽象，对于构建复杂的、多步骤的、带工具调用的工作流非常顺手。它的ConversationBufferMemory、ConversationSummaryMemory等组件可以直接用于管理短期记忆。
• LlamaIndex：优势在于数据连接和检索（RAG）能力超强，其“查询引擎”和“聊天引擎”的概念与我们的“任务线程”模型很契合。它擅长管理复杂的文档上下文，对于需要深度检索外部知识的任务场景更友好。

在这个项目中，我倾向于使用LangChain，因为它对工作流（Workflow）和状态管理的抽象更符合我们的“状态机”思维。我们可以用LLMChain来构建规划器，用AgentExecutor来运行每一步，并用自定义的回调（Callback）函数来在特定节点（如每一步结束后）触发状态保存。

4. 分步实现：从任务创建到中断恢复的全流程

下面，我们进入具体的代码实现环节。我会用伪代码和关键代码片段来说明核心流程，你可以根据自己的环境进行调整。

4.1 第一步：初始化系统与定义数据结构

首先，定义我们的核心数据结构。

# task_state.py from dataclasses import dataclass, asdict from enum import Enum from typing import List, Dict, Any, Optional import json from datetime import datetime class TaskStatus(Enum): CREATED = "created" PLANNING = "planning" EXECUTING = "executing" PAUSED = "paused" COMPLETED = "completed" CANCELLED = "cancelled" @dataclass class TaskStep: step_id: int description: str status: str # "pending", "running", "done", "failed" result: Optional[str] = None @dataclass class Task: task_id: str user_id: str goal: str status: TaskStatus plan: List[TaskStep] # 任务分解后的步骤列表 current_step_index: int # 当前执行到第几步 checkpoint: Dict[str, Any] # 进度快照，存任何需要的信息 created_at: datetime updated_at: datetime def to_dict(self): # 方便序列化存储到数据库 data = asdict(self) data['status'] = self.status.value data['created_at'] = self.created_at.isoformat() data['updated_at'] = self.updated_at.isoformat() return data @classmethod def from_dict(cls, data: Dict[str, Any]): # 从数据库加载 data['status'] = TaskStatus(data['status']) data['created_at'] = datetime.fromisoformat(data['created_at']) data['updated_at'] = datetime.fromisoformat(data['updated_at']) return cls(**data)

然后，初始化我们的核心组件：模型、记忆存储和数据库连接。

# system_init.py import sqlite3 from langchain.llms import Ollama # 假设使用本地Ollama运行Llama3 from langchain.embeddings import HuggingFaceEmbeddings import chromadb # 1. 初始化规划模型（本地） planner_llm = Ollama(model="llama3.1:8b", temperature=0.1) # 低temperature保证规划稳定性 # 2. 初始化执行模型（这里示例用同一个，实践中可用不同的） executor_llm = Ollama(model="llama3.1:8b", temperature=0.7) # 高一点更有创造性 # 3. 初始化嵌入模型和向量数据库（长期记忆） embed_model = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2") # 轻量级嵌入模型 chroma_client = chromadb.PersistentClient(path="./chroma_db") # 注意：我们不会创建一个全局的collection，而是为每个任务动态创建/获取 # 4. 初始化SQLite数据库（任务状态存储） conn = sqlite3.connect('agent_tasks.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks ( task_id TEXT PRIMARY KEY, user_id TEXT, goal TEXT, status TEXT, plan TEXT, -- JSON字符串 current_step_index INTEGER, checkpoint TEXT, -- JSON字符串 created_at TEXT, updated_at TEXT ) ''') conn.commit()

4.2 第二步：实现任务规划与分解模块

这个模块负责接收用户模糊的目标，并将其转化为清晰的步骤列表。

# planner.py from langchain.prompts import PromptTemplate from langchain.chains import LLMChain import json class TaskPlanner: def __init__(self, llm): self.llm = llm self.prompt = PromptTemplate( input_variables=["goal"], template=""" 你是一个高级任务规划AI。请将用户的目标分解成一个清晰、有序、可执行的步骤列表。 每个步骤应该足够具体，以便另一个AI可以独立执行它。 考虑可能需要的迭代或用户确认点。 目标：{goal} 请以严格的JSON数组格式输出，每个元素是一个包含`step_id`(从1开始), `description`(步骤描述)的对象。 示例输出：[{{"step_id": 1, "description": "第一步做什么..."}}, {{"step_id": 2, "description": "第二步做什么..."}}] 输出： """ ) self.chain = LLMChain(llm=self.llm, prompt=self.prompt) def plan(self, goal: str) -> List[TaskStep]: try: result = self.chain.run(goal=goal) # 清理输出，提取JSON部分 json_str = result.strip() if '```json' in json_str: json_str = json_str.split('```json')[1].split('```')[0] elif '```' in json_str: json_str = json_str.split('```')[1].split('```')[0] steps_data = json.loads(json_str) steps = [] for s in steps_data: steps.append(TaskStep(step_id=s['step_id'], description=s['description'], status="pending")) return steps except Exception as e: print(f"规划失败: {e}") # 降级方案：返回一个简单的默认步骤 return [TaskStep(step_id=1, description=f"执行任务：{goal}", status="pending")]

4.3 第三步：构建任务执行引擎与状态保存

这是最核心的部分，它驱动任务一步步前进，并在每一步之后保存状态。

# execution_engine.py from langchain.schema import BaseMessage, HumanMessage, AIMessage from langchain.memory import ConversationBufferMemory import uuid from datetime import datetime class ExecutionEngine: def __init__(self, executor_llm, chroma_client, embed_model, db_conn): self.llm = executor_llm self.chroma_client = chroma_client self.embed_model = embed_model self.db_conn = db_conn self.db_cursor = db_conn.cursor() def create_task(self, user_id: str, goal: str) -> Task: """创建新任务，并初始化规划""" planner = TaskPlanner(planner_llm) # 使用全局的planner_llm steps = planner.plan(goal) task_id = str(uuid.uuid4()) now = datetime.now() task = Task( task_id=task_id, user_id=user_id, goal=goal, status=TaskStatus.CREATED, plan=steps, current_step_index=0, # 还未开始执行任何步骤 checkpoint={"goal": goal, "history": []}, created_at=now, updated_at=now ) # 保存到数据库 self._save_task_to_db(task) # 为这个任务创建专属的向量集合（长期记忆） collection = self.chroma_client.create_collection(name=f"memory_{task_id}") # 初始记忆：任务目标 collection.add( documents=[f"任务目标：{goal}"], metadatas=[{"type": "goal", "step": 0}], ids=[f"goal_{task_id}"] ) task.status = TaskStatus.PLANNING self._update_task_in_db(task) return task def execute_step(self, task_id: str) -> (str, bool): """执行当前步骤，返回执行结果和是否完成标志""" task = self._load_task_from_db(task_id) if task.status != TaskStatus.EXECUTING and task.status != TaskStatus.PLANNING: return f"任务状态为{task.status.value}，无法执行。", False if task.current_step_index >= len(task.plan): task.status = TaskStatus.COMPLETED self._update_task_in_db(task) return "所有步骤已完成！", True current_step = task.plan[task.current_step_index] current_step.status = "running" self._update_task_in_db(task) # 1. 准备上下文：从checkpoint和长期记忆中获取相关信息 context = self._retrieve_relevant_memory(task_id, current_step.description) # 2. 构建给执行模型的提示 prompt = f""" 你正在协助用户完成一个任务。 总目标：{task.goal} 当前步骤（第{current_step.step_id}步）：{current_step.description} 历史上下文： {context} 请执行当前步骤。输出应清晰、完整，直接给出步骤的结果。 输出： """ # 3. 调用模型执行 try: step_result = self.llm.invoke(prompt) except Exception as e: step_result = f"步骤执行出错：{e}" current_step.status = "failed" else: current_step.status = "done" current_step.result = step_result # 4. 保存到长期记忆 collection = self.chroma_client.get_collection(name=f"memory_{task_id}") collection.add( documents=[f"步骤{current_step.step_id}: {current_step.description}\n结果：{step_result}"], metadatas=[{"type": "step_result", "step": current_step.step_id}], ids=[f"step_{current_step.step_id}_{task_id}"] ) # 5. 更新任务状态和checkpoint task.current_step_index += 1 task.checkpoint["last_step"] = current_step.description task.checkpoint["last_result"] = step_result task.checkpoint["history"].append(f"步骤{current_step.step_id}: {step_result[:100]}...") # 存摘要 task.updated_at = datetime.now() if task.current_step_index >= len(task.plan): task.status = TaskStatus.COMPLETED finished = True result_msg = f"步骤 {current_step.step_id} 完成。任务全部完成！\n结果：{step_result}" else: task.status = TaskStatus.EXECUTING finished = False next_step = task.plan[task.current_step_index] result_msg = f"步骤 {current_step.step_id} 完成。\n结果：{step_result}\n\n下一步（步骤{next_step.step_id}）：{next_step.description}" self._update_task_in_db(task) return result_msg, finished def pause_task(self, task_id: str, reason: str = "用户请求中断"): """暂停任务""" task = self._load_task_from_db(task_id) if task.status == TaskStatus.EXECUTING: task.status = TaskStatus.PAUSED task.checkpoint["pause_reason"] = reason task.updated_at = datetime.now() self._update_task_in_db(task) # 在长期记忆中也记录一次中断 collection = self.chroma_client.get_collection(name=f"memory_{task_id}") collection.add( documents=[f"任务于{datetime.now()}被暂停。原因：{reason}"], metadatas=[{"type": "system_event", "step": "pause"}], ids=[f"pause_{task_id}_{datetime.now().timestamp()}"] ) return True return False def resume_task(self, task_id: str) -> str: """恢复被暂停的任务""" task = self._load_task_from_db(task_id) if task.status == TaskStatus.PAUSED: task.status = TaskStatus.EXECUTING task.updated_at = datetime.now() self._update_task_in_db(task) # 生成恢复提示 last_step_info = "" if task.current_step_index > 0 and task.current_step_index <= len(task.plan): last_step = task.plan[task.current_step_index - 1] last_step_info = f"我们刚刚完成了：{last_step.description}\n结果摘要：{last_step.result[:150]}...\n" next_step = task.plan[task.current_step_index] if task.current_step_index < len(task.plan) else None next_step_info = f"接下来继续：{next_step.description}" if next_step else "所有步骤已完成。" resume_msg = f"任务已恢复。{last_step_info}{next_step_info}" return resume_msg else: return f"任务当前状态为{task.status.value}，无法恢复。" def _retrieve_relevant_memory(self, task_id: str, query: str, n_results: int = 3) -> str: """从该任务的长期记忆中检索相关信息""" try: collection = self.chroma_client.get_collection(name=f"memory_{task_id}") results = collection.query( query_texts=[query], n_results=n_results ) if results and results['documents']: return "\n".join(results['documents'][0]) except Exception as e: print(f"记忆检索失败: {e}") return "（暂无相关历史记忆）" def _save_task_to_db(self, task: Task): # ... (实现数据库插入逻辑) pass def _load_task_from_db(self, task_id: str) -> Task: # ... (实现数据库查询逻辑) pass def _update_task_in_db(self, task: Task): # ... (实现数据库更新逻辑) pass

4.4 第四步：设计主控循环与用户交互接口

最后，我们需要一个主循环来连接用户输入和我们的引擎。

# main_controller.py import re class AgentController: def __init__(self, engine: ExecutionEngine): self.engine = engine self.active_tasks = {} # user_id -> task_id 的映射，简化处理，实际应存数据库 def process_input(self, user_id: str, user_input: str) -> str: # 1. 意图识别：是新建任务、控制指令还是继续对话？ if user_input.lower().startswith("/new "): goal = user_input[5:].strip() task = self.engine.create_task(user_id, goal) self.active_tasks[user_id] = task.task_id # 开始执行第一步 task.status = TaskStatus.EXECUTING self.engine._update_task_in_db(task) result, finished = self.engine.execute_step(task.task_id) return f"已创建任务【{goal}】。\n{result}" elif user_input.lower().startswith("/pause"): if user_id in self.active_tasks: task_id = self.active_tasks[user_id] if self.engine.pause_task(task_id): return "任务已暂停。你可以随时输入 /resume 来恢复。" else: return "当前没有正在执行的任务可暂停。" return "你当前没有活跃任务。" elif user_input.lower().startswith("/resume"): if user_id in self.active_tasks: task_id = self.active_tasks[user_id] resume_msg = self.engine.resume_task(task_id) # 恢复后自动执行下一步 result, finished = self.engine.execute_step(task_id) return f"{resume_msg}\n\n{result}" return "你当前没有可恢复的任务。请使用 /new 创建一个新任务。" elif user_input.lower().startswith("/list"): # 查询该用户的所有任务（简化示例） return self._list_user_tasks(user_id) # 2. 如果是普通输入，且当前有活跃任务，则视为对当前步骤的补充或反馈 elif user_id in self.active_tasks: task_id = self.active_tasks[user_id] task = self.engine._load_task_from_db(task_id) # 将用户输入作为额外信息存入记忆，可能影响下一步执行 collection = self.engine.chroma_client.get_collection(name=f"memory_{task_id}") collection.add( documents=[f"用户额外输入：{user_input}"], metadatas=[{"type": "user_feedback", "step": task.current_step_index}], ids=[f"feedback_{task_id}_{datetime.now().timestamp()}"] ) # 然后继续执行下一步（或者根据设计，可以重新执行当前步） result, finished = self.engine.execute_step(task_id) return f"已记录你的反馈。\n{result}" # 3. 其他情况：当作新任务的开始（简化处理） else: return "似乎你想开始一个新任务？请使用 /new [你的任务目标] 的格式告诉我，例如：/new 帮我写一封英文会议邀请函" def _list_user_tasks(self, user_id: str) -> str: # 查询数据库并格式化输出 # ... (实现数据库查询) return "你的任务列表：\n1. [任务ID] 写周报 (状态：进行中)\n2. [任务ID] 策划活动 (状态：已暂停)"

5. 常见问题、调试技巧与优化方向

在实际搭建和测试过程中，我遇到了不少坑，也总结出一些让Agent更“聪明”的技巧。

5.1 模型不按格式输出怎么办？

这是使用开源模型最常见的问题。规划步骤时，我们要求模型输出JSON，但它可能返回一段自由文本。

解决方案：

1. 强化提示词（Prompt Engineering）：在提示词中明确给出格式，并使用“json ...”这样的标记来框定输出范围。示例中我们已经这样做了。
2. 后处理解析：像代码中那样，尝试提取两个“```”之间的内容，或者寻找第一个“[”和最后一个“]”之间的内容。可以写一个健壮的解析函数，尝试json.loads()，如果失败，就用正则表达式清理文本再试。
3. 使用输出解析器（Output Parser）：LangChain提供了PydanticOutputParser、StructuredOutputParser等工具，能强制模型按指定格式输出，大大降低解析失败率。这是更优雅的解决方案。

5.2 任务分解不合理或步骤太粗/太细？

模型可能把“写报告”分解成“打开电脑”、“打开Word”、“开始写”这样无意义的步骤，或者相反，步骤过于笼统。

解决方案：

1. 提供示例（Few-shot Prompting）：在给规划模型的提示词中，包含1-2个高质量的任务分解示例。例如：“目标：组织一次线上技术分享会。输出示例：[{"step_id":1, "description":"确定分享主题和核心听众"}, {"step_id":2, "description":"邀请2-3位潜在的分享嘉宾并协调时间"}, ...]”
2. 迭代规划：不要一次性生成所有步骤。可以先让模型生成一个高层大纲（如：1. 准备阶段 2. 执行阶段 3. 收尾阶段），然后对每个阶段再分别进行细化分解。这更符合人类的规划习惯。
3. 人工审核介入点：在规划提示词中明确：“在涉及关键决策（如预算分配、技术选型）或需要外部信息（如查询某产品价格）的步骤后，应标记为need_human_input。”这样，Agent会在这些节点主动暂停，等待用户输入。

5.3 长期记忆检索不准，导致上下文混乱

向量检索并非总是精准，可能召回不相关的记忆片段，干扰模型判断。

解决方案：

1. 优化记忆片段（Chunking）：存入向量数据库的文本不宜过长或过短。一个步骤的结果可以作为一个片段。对于长文档，可以按语义段落切割。好的分块策略能极大提升检索质量。
2. 混合检索（Hybrid Search）：不要只依赖向量相似度搜索。可以结合关键词搜索（如BM25）。ChromaDB等数据库支持混合检索。对于确切的名称、日期、数字，关键词搜索往往更准。
3. 重排序（Re-ranking）：先召回较多的候选片段（比如10个），再用一个更小、更快的交叉编码器（Cross-Encoder）模型对它们进行相关性重排序，只取Top 3给模型。这能显著提升精度，但会增加计算开销。
4. 记忆元数据过滤：在检索时，利用我们存入的元数据（如{"type": "step_result", "step": 2}）进行过滤。例如，当执行第5步时，我们可以优先检索step为4或5的记忆，而不是所有记忆。

5.4 如何评估Agent的表现？

除了人工测试，可以建立一些自动化评估指标：

• 任务完成率：给定10个标准任务，有多少个能从头到尾无需人工干预（除了设计好的介入点）地走完流程？
• 中断恢复成功率：在任意步骤暂停后，恢复指令能否正确载入上下文并继续？恢复后的下一步执行是否正确？
• 步骤合理性人工评分：请多人对分解出的步骤进行1-5分打分，评估其逻辑性和可执行性。
• 幻觉率：在需要事实性知识的步骤中（如“查询2023年全球智能手机出货量”），Agent是否编造了不存在的数据？

搭建这个AI Agent的过程，更像是在设计一套人机交互的“协议”和“工作流”。技术本身（模型、向量数据库）是工具，而如何让这些工具流畅地协作，理解并适应人类非线性、多线程的工作方式，才是真正的挑战。这个原型仅仅是一个起点，你可以在此基础上增加更复杂的特性，比如多任务并行管理、工具调用（让Agent能自己上网搜索、操作软件）、甚至让多个Agent之间进行协作。希望这个详细的拆解能给你提供一个坚实的起点，去创造真正懂你、能与你并肩工作的AI伙伴。