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伙伴。