1. 项目概述为什么我们需要一个能“被打断”的AI助手最近在折腾一个挺有意思的项目核心目标就写在标题里了从零开始搭建一个能理解“人机协作”与“中断恢复”的AI Agent。这听起来可能有点抽象但如果你用过市面上那些“一问一答”式的聊天机器人肯定遇到过这样的场景你正在让它帮你写一份报告刚列好大纲老板突然在群里你让你立刻处理一个紧急数据。你不得不中断和AI的对话等忙完回来要么得从头开始描述你的需求要么AI已经“失忆”完全不记得刚才的上下文了。这就是传统AI交互模式的痛点——它假设对话是线性的、连续的但真实世界的工作流充满了中断、并行和优先级切换。一个真正有用的AI助手不应该只是一个更聪明的搜索引擎而应该像一个得力的工作伙伴能理解任务的上下文在你离开后能“暂停”在你回来时能“续上”甚至能主动管理多个并行的任务线程。这就是我动手搭建这个AI Agent的初衷探索如何让AI具备基础的“工作记忆”和“任务状态管理”能力实现更自然、更高效的人机协作。这个项目不依赖于任何单一的、庞大的语言模型而是侧重于架构设计与流程控制。我们会用一些开源的轻量级模型和框架作为核心组件但更重要的是构建一套让这些组件协同工作的逻辑。最终实现的Agent将能够接受一个复杂任务比如“帮我策划一次团队outing”将其分解为多个步骤定主题、选地点、做预算、发通知并允许你在任何步骤介入、修改或暂停Agent会记住所有状态确保任务最终被完整、正确地执行。2. 核心设计思路将“协作”与“中断”机制化2.1 打破线性对话任务即状态机第一个要摒弃的观念就是把AI对话看作一连串的QA。在我们的设计中每一个用户发起的顶层任务都被实例化为一个独立的状态机State Machine。这个状态机有明确的生命周期创建 - 规划 - 执行 - 暂停 - 恢复 - 完成/取消。为什么是状态机因为它天然适合描述有步骤、可中断的过程。比如“写周报”这个任务其状态可能包括等待输入原始数据、生成初稿、等待用户审阅、根据反馈修改、最终定稿。当用户说“等一下我先发个邮件”Agent不应该清空“写周报”的所有中间数据而仅仅是将状态从执行切换到暂停并持久化保存当前的进度比如已经生成的初稿内容、收集到的数据点。注意这里的状态持久化是关键。你不能只把状态存在内存里一旦服务重启就全丢了。最简单的做法是使用一个轻量级数据库如SQLite或甚至一个JSON文件来记录每个任务ID对应的状态快照snapshot。快照里需要包含任务目标、已完成的步骤、当前的输出、下一步的指令、以及相关的上下文数据。2.2 构建智能体的“工作记忆”上下文管理与会话线程要让AI记住“刚才说到哪了”我们需要设计一个分层的上下文管理系统。这不仅仅是把最近的几条对话记录塞给模型那么简单。会话线程Conversation Thread每个独立的任务就是一个线程。用户和AI围绕这个任务的所有交互都归属于这个线程。线程ID是唯一的便于检索和恢复。短期工作记忆Short-term Working Memory这是模型上下文窗口比如GPT的16K、32K tokens直接能“看到”的内容。我们只在这里存放最相关、最精简的信息例如当前步骤的指令、上一步的输出、以及关键的决策历史。目的是节省宝贵的Token并提高模型对当前任务的专注度。长期记忆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成本高且不利于深度定制。我采用混合模式规划与调度模型Planner选用Llama 3.1 8B或Qwen 2.5 7B这类中等规模的开源模型。它们的推理能力足够强可以理解复杂任务并将其分解成清晰的步骤列表Step-by-step Plan。这个模型负责“想”即任务分解和步骤规划。我们可以使用LM Studio或Ollama在本地运行它响应快且隐私性好。执行与对话模型Executor对于每一步的具体执行比如根据大纲写一段文字、写一段Python代码、或者回答用户的具体问题可以视情况选择。对质量要求高的步骤可以调用DeepSeek-V3或GPT-3.5-Turbo的API对简单、格式化的响应可以用更小的本地模型如Phi-3-mini。这个模型负责“做”即执行具体指令。摘要与工具调用模型Helper用于生成对话摘要、判断用户意图是提问、指令还是中断信号、以及决定是否需要调用外部工具如计算器、搜索引擎。这个角色对能力要求相对较低可以使用非常轻量的模型如Gemma 2B以降低整体延迟和资源消耗。为什么这么选将“规划”和“执行”分离符合人类的思考方式也使得系统更模块化、更易调试。规划模型一次生成整个计划执行模型则专注于当前步骤互不干扰。同时混合使用本地模型和云端API在成本、性能和隐私之间取得了平衡。3.2 记忆系统向量数据库与结构化存储向量数据库长期记忆我选择ChromaDB。它轻量、易用Python集成度极高非常适合原型和中小型项目。我们将每个任务线程的所有“记忆片段”用户输入、AI输出、生成的文档块都通过text-embedding-3-small这类嵌入模型转换成向量存入ChromaDB的对应集合Collection中集合名可以用任务ID命名。结构化存储任务状态使用SQLite足矣。我们需要一张tasks表核心字段包括task_id(主键)user_id(区分不同用户)goal(任务目标)current_state(如 “planning”, “executing_step_3”, “paused”)plan(JSON格式存储的任务分解步骤)checkpoint(JSON格式存储的进度快照如上一步输出)created_at,updated_at3.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(modelllama3.1:8b, temperature0.1) # 低temperature保证规划稳定性 # 2. 初始化执行模型这里示例用同一个实践中可用不同的 executor_llm Ollama(modelllama3.1:8b, temperature0.7) # 高一点更有创造性 # 3. 初始化嵌入模型和向量数据库长期记忆 embed_model HuggingFaceEmbeddings(model_nameall-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(llmself.llm, promptself.prompt) def plan(self, goal: str) - List[TaskStep]: try: result self.chain.run(goalgoal) # 清理输出提取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_ids[step_id], descriptions[description], statuspending)) return steps except Exception as e: print(f规划失败: {e}) # 降级方案返回一个简单的默认步骤 return [TaskStep(step_id1, descriptionf执行任务{goal}, statuspending)]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_idtask_id, user_iduser_id, goalgoal, statusTaskStatus.CREATED, plansteps, current_step_index0, # 还未开始执行任何步骤 checkpoint{goal: goal, history: []}, created_atnow, updated_atnow ) # 保存到数据库 self._save_task_to_db(task) # 为这个任务创建专属的向量集合长期记忆 collection self.chroma_client.create_collection(namefmemory_{task_id}) # 初始记忆任务目标 collection.add( documents[f任务目标{goal}], metadatas[{type: goal, step: 0}], ids[fgoal_{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(namefmemory_{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[fstep_{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(namefmemory_{task_id}) collection.add( documents[f任务于{datetime.now()}被暂停。原因{reason}], metadatas[{type: system_event, step: pause}], ids[fpause_{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(namefmemory_{task_id}) results collection.query( query_texts[query], n_resultsn_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): # ... (实现数据库更新逻辑) pass4.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(namefmemory_{task_id}) collection.add( documents[f用户额外输入{user_input}], metadatas[{type: user_feedback, step: task.current_step_index}], ids[ffeedback_{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但它可能返回一段自由文本。解决方案强化提示词Prompt Engineering在提示词中明确给出格式并使用“json ...”这样的标记来框定输出范围。示例中我们已经这样做了。后处理解析像代码中那样尝试提取两个“”之间的内容或者寻找第一个“[”和最后一个“]”之间的内容。可以写一个健壮的解析函数尝试json.loads()如果失败就用正则表达式清理文本再试。使用输出解析器Output ParserLangChain提供了PydanticOutputParser、StructuredOutputParser等工具能强制模型按指定格式输出大大降低解析失败率。这是更优雅的解决方案。5.2 任务分解不合理或步骤太粗/太细模型可能把“写报告”分解成“打开电脑”、“打开Word”、“开始写”这样无意义的步骤或者相反步骤过于笼统。解决方案提供示例Few-shot Prompting在给规划模型的提示词中包含1-2个高质量的任务分解示例。例如“目标组织一次线上技术分享会。输出示例[{step_id:1, description:确定分享主题和核心听众}, {step_id:2, description:邀请2-3位潜在的分享嘉宾并协调时间}, ...]”迭代规划不要一次性生成所有步骤。可以先让模型生成一个高层大纲如1. 准备阶段 2. 执行阶段 3. 收尾阶段然后对每个阶段再分别进行细化分解。这更符合人类的规划习惯。人工审核介入点在规划提示词中明确“在涉及关键决策如预算分配、技术选型或需要外部信息如查询某产品价格的步骤后应标记为need_human_input。”这样Agent会在这些节点主动暂停等待用户输入。5.3 长期记忆检索不准导致上下文混乱向量检索并非总是精准可能召回不相关的记忆片段干扰模型判断。解决方案优化记忆片段Chunking存入向量数据库的文本不宜过长或过短。一个步骤的结果可以作为一个片段。对于长文档可以按语义段落切割。好的分块策略能极大提升检索质量。混合检索Hybrid Search不要只依赖向量相似度搜索。可以结合关键词搜索如BM25。ChromaDB等数据库支持混合检索。对于确切的名称、日期、数字关键词搜索往往更准。重排序Re-ranking先召回较多的候选片段比如10个再用一个更小、更快的交叉编码器Cross-Encoder模型对它们进行相关性重排序只取Top 3给模型。这能显著提升精度但会增加计算开销。记忆元数据过滤在检索时利用我们存入的元数据如{type: step_result, step: 2}进行过滤。例如当执行第5步时我们可以优先检索step为4或5的记忆而不是所有记忆。5.4 如何评估Agent的表现除了人工测试可以建立一些自动化评估指标任务完成率给定10个标准任务有多少个能从头到尾无需人工干预除了设计好的介入点地走完流程中断恢复成功率在任意步骤暂停后恢复指令能否正确载入上下文并继续恢复后的下一步执行是否正确步骤合理性人工评分请多人对分解出的步骤进行1-5分打分评估其逻辑性和可执行性。幻觉率在需要事实性知识的步骤中如“查询2023年全球智能手机出货量”Agent是否编造了不存在的数据搭建这个AI Agent的过程更像是在设计一套人机交互的“协议”和“工作流”。技术本身模型、向量数据库是工具而如何让这些工具流畅地协作理解并适应人类非线性、多线程的工作方式才是真正的挑战。这个原型仅仅是一个起点你可以在此基础上增加更复杂的特性比如多任务并行管理、工具调用让Agent能自己上网搜索、操作软件、甚至让多个Agent之间进行协作。希望这个详细的拆解能给你提供一个坚实的起点去创造真正懂你、能与你并肩工作的AI伙伴。 从零构建支持中断恢复的AI Agent：状态机与记忆系统设计实践
1. 项目概述为什么我们需要一个能“被打断”的AI助手最近在折腾一个挺有意思的项目核心目标就写在标题里了从零开始搭建一个能理解“人机协作”与“中断恢复”的AI Agent。这听起来可能有点抽象但如果你用过市面上那些“一问一答”式的聊天机器人肯定遇到过这样的场景你正在让它帮你写一份报告刚列好大纲老板突然在群里你让你立刻处理一个紧急数据。你不得不中断和AI的对话等忙完回来要么得从头开始描述你的需求要么AI已经“失忆”完全不记得刚才的上下文了。这就是传统AI交互模式的痛点——它假设对话是线性的、连续的但真实世界的工作流充满了中断、并行和优先级切换。一个真正有用的AI助手不应该只是一个更聪明的搜索引擎而应该像一个得力的工作伙伴能理解任务的上下文在你离开后能“暂停”在你回来时能“续上”甚至能主动管理多个并行的任务线程。这就是我动手搭建这个AI Agent的初衷探索如何让AI具备基础的“工作记忆”和“任务状态管理”能力实现更自然、更高效的人机协作。这个项目不依赖于任何单一的、庞大的语言模型而是侧重于架构设计与流程控制。我们会用一些开源的轻量级模型和框架作为核心组件但更重要的是构建一套让这些组件协同工作的逻辑。最终实现的Agent将能够接受一个复杂任务比如“帮我策划一次团队outing”将其分解为多个步骤定主题、选地点、做预算、发通知并允许你在任何步骤介入、修改或暂停Agent会记住所有状态确保任务最终被完整、正确地执行。2. 核心设计思路将“协作”与“中断”机制化2.1 打破线性对话任务即状态机第一个要摒弃的观念就是把AI对话看作一连串的QA。在我们的设计中每一个用户发起的顶层任务都被实例化为一个独立的状态机State Machine。这个状态机有明确的生命周期创建 - 规划 - 执行 - 暂停 - 恢复 - 完成/取消。为什么是状态机因为它天然适合描述有步骤、可中断的过程。比如“写周报”这个任务其状态可能包括等待输入原始数据、生成初稿、等待用户审阅、根据反馈修改、最终定稿。当用户说“等一下我先发个邮件”Agent不应该清空“写周报”的所有中间数据而仅仅是将状态从执行切换到暂停并持久化保存当前的进度比如已经生成的初稿内容、收集到的数据点。注意这里的状态持久化是关键。你不能只把状态存在内存里一旦服务重启就全丢了。最简单的做法是使用一个轻量级数据库如SQLite或甚至一个JSON文件来记录每个任务ID对应的状态快照snapshot。快照里需要包含任务目标、已完成的步骤、当前的输出、下一步的指令、以及相关的上下文数据。2.2 构建智能体的“工作记忆”上下文管理与会话线程要让AI记住“刚才说到哪了”我们需要设计一个分层的上下文管理系统。这不仅仅是把最近的几条对话记录塞给模型那么简单。会话线程Conversation Thread每个独立的任务就是一个线程。用户和AI围绕这个任务的所有交互都归属于这个线程。线程ID是唯一的便于检索和恢复。短期工作记忆Short-term Working Memory这是模型上下文窗口比如GPT的16K、32K tokens直接能“看到”的内容。我们只在这里存放最相关、最精简的信息例如当前步骤的指令、上一步的输出、以及关键的决策历史。目的是节省宝贵的Token并提高模型对当前任务的专注度。长期记忆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成本高且不利于深度定制。我采用混合模式规划与调度模型Planner选用Llama 3.1 8B或Qwen 2.5 7B这类中等规模的开源模型。它们的推理能力足够强可以理解复杂任务并将其分解成清晰的步骤列表Step-by-step Plan。这个模型负责“想”即任务分解和步骤规划。我们可以使用LM Studio或Ollama在本地运行它响应快且隐私性好。执行与对话模型Executor对于每一步的具体执行比如根据大纲写一段文字、写一段Python代码、或者回答用户的具体问题可以视情况选择。对质量要求高的步骤可以调用DeepSeek-V3或GPT-3.5-Turbo的API对简单、格式化的响应可以用更小的本地模型如Phi-3-mini。这个模型负责“做”即执行具体指令。摘要与工具调用模型Helper用于生成对话摘要、判断用户意图是提问、指令还是中断信号、以及决定是否需要调用外部工具如计算器、搜索引擎。这个角色对能力要求相对较低可以使用非常轻量的模型如Gemma 2B以降低整体延迟和资源消耗。为什么这么选将“规划”和“执行”分离符合人类的思考方式也使得系统更模块化、更易调试。规划模型一次生成整个计划执行模型则专注于当前步骤互不干扰。同时混合使用本地模型和云端API在成本、性能和隐私之间取得了平衡。3.2 记忆系统向量数据库与结构化存储向量数据库长期记忆我选择ChromaDB。它轻量、易用Python集成度极高非常适合原型和中小型项目。我们将每个任务线程的所有“记忆片段”用户输入、AI输出、生成的文档块都通过text-embedding-3-small这类嵌入模型转换成向量存入ChromaDB的对应集合Collection中集合名可以用任务ID命名。结构化存储任务状态使用SQLite足矣。我们需要一张tasks表核心字段包括task_id(主键)user_id(区分不同用户)goal(任务目标)current_state(如 “planning”, “executing_step_3”, “paused”)plan(JSON格式存储的任务分解步骤)checkpoint(JSON格式存储的进度快照如上一步输出)created_at,updated_at3.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(modelllama3.1:8b, temperature0.1) # 低temperature保证规划稳定性 # 2. 初始化执行模型这里示例用同一个实践中可用不同的 executor_llm Ollama(modelllama3.1:8b, temperature0.7) # 高一点更有创造性 # 3. 初始化嵌入模型和向量数据库长期记忆 embed_model HuggingFaceEmbeddings(model_nameall-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(llmself.llm, promptself.prompt) def plan(self, goal: str) - List[TaskStep]: try: result self.chain.run(goalgoal) # 清理输出提取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_ids[step_id], descriptions[description], statuspending)) return steps except Exception as e: print(f规划失败: {e}) # 降级方案返回一个简单的默认步骤 return [TaskStep(step_id1, descriptionf执行任务{goal}, statuspending)]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_idtask_id, user_iduser_id, goalgoal, statusTaskStatus.CREATED, plansteps, current_step_index0, # 还未开始执行任何步骤 checkpoint{goal: goal, history: []}, created_atnow, updated_atnow ) # 保存到数据库 self._save_task_to_db(task) # 为这个任务创建专属的向量集合长期记忆 collection self.chroma_client.create_collection(namefmemory_{task_id}) # 初始记忆任务目标 collection.add( documents[f任务目标{goal}], metadatas[{type: goal, step: 0}], ids[fgoal_{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(namefmemory_{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[fstep_{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(namefmemory_{task_id}) collection.add( documents[f任务于{datetime.now()}被暂停。原因{reason}], metadatas[{type: system_event, step: pause}], ids[fpause_{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(namefmemory_{task_id}) results collection.query( query_texts[query], n_resultsn_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): # ... (实现数据库更新逻辑) pass4.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(namefmemory_{task_id}) collection.add( documents[f用户额外输入{user_input}], metadatas[{type: user_feedback, step: task.current_step_index}], ids[ffeedback_{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但它可能返回一段自由文本。解决方案强化提示词Prompt Engineering在提示词中明确给出格式并使用“json ...”这样的标记来框定输出范围。示例中我们已经这样做了。后处理解析像代码中那样尝试提取两个“”之间的内容或者寻找第一个“[”和最后一个“]”之间的内容。可以写一个健壮的解析函数尝试json.loads()如果失败就用正则表达式清理文本再试。使用输出解析器Output ParserLangChain提供了PydanticOutputParser、StructuredOutputParser等工具能强制模型按指定格式输出大大降低解析失败率。这是更优雅的解决方案。5.2 任务分解不合理或步骤太粗/太细模型可能把“写报告”分解成“打开电脑”、“打开Word”、“开始写”这样无意义的步骤或者相反步骤过于笼统。解决方案提供示例Few-shot Prompting在给规划模型的提示词中包含1-2个高质量的任务分解示例。例如“目标组织一次线上技术分享会。输出示例[{step_id:1, description:确定分享主题和核心听众}, {step_id:2, description:邀请2-3位潜在的分享嘉宾并协调时间}, ...]”迭代规划不要一次性生成所有步骤。可以先让模型生成一个高层大纲如1. 准备阶段 2. 执行阶段 3. 收尾阶段然后对每个阶段再分别进行细化分解。这更符合人类的规划习惯。人工审核介入点在规划提示词中明确“在涉及关键决策如预算分配、技术选型或需要外部信息如查询某产品价格的步骤后应标记为need_human_input。”这样Agent会在这些节点主动暂停等待用户输入。5.3 长期记忆检索不准导致上下文混乱向量检索并非总是精准可能召回不相关的记忆片段干扰模型判断。解决方案优化记忆片段Chunking存入向量数据库的文本不宜过长或过短。一个步骤的结果可以作为一个片段。对于长文档可以按语义段落切割。好的分块策略能极大提升检索质量。混合检索Hybrid Search不要只依赖向量相似度搜索。可以结合关键词搜索如BM25。ChromaDB等数据库支持混合检索。对于确切的名称、日期、数字关键词搜索往往更准。重排序Re-ranking先召回较多的候选片段比如10个再用一个更小、更快的交叉编码器Cross-Encoder模型对它们进行相关性重排序只取Top 3给模型。这能显著提升精度但会增加计算开销。记忆元数据过滤在检索时利用我们存入的元数据如{type: step_result, step: 2}进行过滤。例如当执行第5步时我们可以优先检索step为4或5的记忆而不是所有记忆。5.4 如何评估Agent的表现除了人工测试可以建立一些自动化评估指标任务完成率给定10个标准任务有多少个能从头到尾无需人工干预除了设计好的介入点地走完流程中断恢复成功率在任意步骤暂停后恢复指令能否正确载入上下文并继续恢复后的下一步执行是否正确步骤合理性人工评分请多人对分解出的步骤进行1-5分打分评估其逻辑性和可执行性。幻觉率在需要事实性知识的步骤中如“查询2023年全球智能手机出货量”Agent是否编造了不存在的数据搭建这个AI Agent的过程更像是在设计一套人机交互的“协议”和“工作流”。技术本身模型、向量数据库是工具而如何让这些工具流畅地协作理解并适应人类非线性、多线程的工作方式才是真正的挑战。这个原型仅仅是一个起点你可以在此基础上增加更复杂的特性比如多任务并行管理、工具调用让Agent能自己上网搜索、操作软件、甚至让多个Agent之间进行协作。希望这个详细的拆解能给你提供一个坚实的起点去创造真正懂你、能与你并肩工作的AI伙伴。