1. 項目概述為什么我們需要一個能“被打斷”的AI助手最近在折騰一個挺有意思的項目核心目標(biāo)就寫在標(biāo)題里了從零開始搭建一個能理解“人機協(xié)作”與“中斷恢復(fù)”的AI Agent。這聽起來可能有點抽象但如果你用過市面上那些“一問一答”式的聊天機器人肯定遇到過這樣的場景你正在讓它幫你寫一份報告剛列好大綱老板突然在群里你讓你立刻處理一個緊急數(shù)據(jù)。你不得不中斷和AI的對話等忙完回來要么得從頭開始描述你的需求要么AI已經(jīng)“失憶”完全不記得剛才的上下文了。這就是傳統(tǒng)AI交互模式的痛點——它假設(shè)對話是線性的、連續(xù)的但真實世界的工作流充滿了中斷、并行和優(yōu)先級切換。一個真正有用的AI助手不應(yīng)該只是一個更聰明的搜索引擎而應(yīng)該像一個得力的工作伙伴能理解任務(wù)的上下文在你離開后能“暫停”在你回來時能“續(xù)上”甚至能主動管理多個并行的任務(wù)線程。這就是我動手搭建這個AI Agent的初衷探索如何讓AI具備基礎(chǔ)的“工作記憶”和“任務(wù)狀態(tài)管理”能力實現(xiàn)更自然、更高效的人機協(xié)作。這個項目不依賴于任何單一的、龐大的語言模型而是側(cè)重于架構(gòu)設(shè)計與流程控制。我們會用一些開源的輕量級模型和框架作為核心組件但更重要的是構(gòu)建一套讓這些組件協(xié)同工作的邏輯。最終實現(xiàn)的Agent將能夠接受一個復(fù)雜任務(wù)比如“幫我策劃一次團隊outing”將其分解為多個步驟定主題、選地點、做預(yù)算、發(fā)通知并允許你在任何步驟介入、修改或暫停Agent會記住所有狀態(tài)確保任務(wù)最終被完整、正確地執(zhí)行。2. 核心設(shè)計思路將“協(xié)作”與“中斷”機制化2.1 打破線性對話任務(wù)即狀態(tài)機第一個要摒棄的觀念就是把AI對話看作一連串的QA。在我們的設(shè)計中每一個用戶發(fā)起的頂層任務(wù)都被實例化為一個獨立的狀態(tài)機State Machine。這個狀態(tài)機有明確的生命周期創(chuàng)建 - 規(guī)劃 - 執(zhí)行 - 暫停 - 恢復(fù) - 完成/取消。為什么是狀態(tài)機因為它天然適合描述有步驟、可中斷的過程。比如“寫周報”這個任務(wù)其狀態(tài)可能包括等待輸入原始數(shù)據(jù)、生成初稿、等待用戶審閱、根據(jù)反饋修改、最終定稿。當(dāng)用戶說“等一下我先發(fā)個郵件”Agent不應(yīng)該清空“寫周報”的所有中間數(shù)據(jù)而僅僅是將狀態(tài)從執(zhí)行切換到暫停并持久化保存當(dāng)前的進度比如已經(jīng)生成的初稿內(nèi)容、收集到的數(shù)據(jù)點。注意這里的狀態(tài)持久化是關(guān)鍵。你不能只把狀態(tài)存在內(nèi)存里一旦服務(wù)重啟就全丟了。最簡單的做法是使用一個輕量級數(shù)據(jù)庫如SQLite或甚至一個JSON文件來記錄每個任務(wù)ID對應(yīng)的狀態(tài)快照snapshot。快照里需要包含任務(wù)目標(biāo)、已完成的步驟、當(dāng)前的輸出、下一步的指令、以及相關(guān)的上下文數(shù)據(jù)。2.2 構(gòu)建智能體的“工作記憶”上下文管理與會話線程要讓AI記住“剛才說到哪了”我們需要設(shè)計一個分層的上下文管理系統(tǒng)。這不僅僅是把最近的幾條對話記錄塞給模型那么簡單。會話線程Conversation Thread每個獨立的任務(wù)就是一個線程。用戶和AI圍繞這個任務(wù)的所有交互都歸屬于這個線程。線程ID是唯一的便于檢索和恢復(fù)。短期工作記憶Short-term Working Memory這是模型上下文窗口比如GPT的16K、32K tokens直接能“看到”的內(nèi)容。我們只在這里存放最相關(guān)、最精簡的信息例如當(dāng)前步驟的指令、上一步的輸出、以及關(guān)鍵的決策歷史。目的是節(jié)省寶貴的Token并提高模型對當(dāng)前任務(wù)的專注度。長期記憶Long-term Memory這是存儲在外部向量數(shù)據(jù)庫如ChromaDB, FAISS里的東西。當(dāng)對話進行時我們把重要的中間結(jié)論、用戶提供的詳細資料、生成的文檔片段轉(zhuǎn)換成向量Embeddings存起來。當(dāng)Agent需要“回憶”某個細節(jié)時比如用戶問“剛才我們定的預(yù)算是多少”它可以通過語義搜索從長期記憶中快速檢索出相關(guān)片段再注入到短期記憶中供模型使用。實操心得很多新手會犯一個錯誤就是把整個冗長的對話歷史都扔進短期記憶。這會導(dǎo)致模型注意力分散并且很快耗盡上下文長度。正確的做法是進行摘要Summarization。每完成一個任務(wù)步驟或經(jīng)過一定輪次的對話就用一個小模型或調(diào)用大模型的摘要功能對當(dāng)前進展做一次摘要用這個摘要來更新短期記憶的“頭部”而將原始細節(jié)移入長期記憶。這樣既能保持連貫性又能控制上下文長度。2.3 設(shè)計中斷與恢復(fù)協(xié)議明確的信號與清晰的邊界中斷不是錯誤而是一種正常的操作指令。因此我們需要為Agent定義一套清晰的中斷與恢復(fù)協(xié)議。中斷信號用戶可以通過特定指令如“/pause”、“稍等我先處理點別的”或簡單地開啟一個全新主題的對話來發(fā)出中斷信號。Agent需要能識別這些信號。狀態(tài)保存點Checkpoint收到中斷信號后Agent應(yīng)立即觸發(fā)一個“保存點”操作。這包括將當(dāng)前任務(wù)狀態(tài)機的狀態(tài)序列化保存。將短期工作記憶中的關(guān)鍵上下文進行摘要并保存。記錄下導(dǎo)致中斷的用戶查詢?nèi)绻械脑捵鳛榛謴?fù)時的參考。恢復(fù)機制當(dāng)用戶回來通過指令如“/resume [任務(wù)ID]”、“繼續(xù)剛才寫周報的事”或語義匹配用戶問“我們剛才說到哪了”要求恢復(fù)時Agent需要根據(jù)任務(wù)ID或語義搜索找到對應(yīng)的任務(wù)上下文。加載保存的狀態(tài)和記憶摘要。生成一條友好的恢復(fù)提示例如“歡迎回來我們剛才正在‘撰寫Q2項目復(fù)盤周報’。我已經(jīng)完成了背景部分和主要成就的初稿接下來準(zhǔn)備寫‘遇到的挑戰(zhàn)與改進’。我們從這里繼續(xù)好嗎”這套協(xié)議的核心是讓中斷和恢復(fù)變得可預(yù)測、可管理而不是一次意外的對話崩潰。3. 技術(shù)棧選型與核心模塊拆解搭建這樣一個Agent我們不需要從頭造輪子合理利用現(xiàn)有開源生態(tài)是關(guān)鍵。以下是我的技術(shù)選型思路兼顧了能力、復(fù)雜度和學(xué)習(xí)成本。3.1 大腦核心輕量級與功能型模型搭配完全依賴GPT-4這樣的大型商用API成本高且不利于深度定制。我采用混合模式規(guī)劃與調(diào)度模型Planner選用Llama 3.1 8B或Qwen 2.5 7B這類中等規(guī)模的開源模型。它們的推理能力足夠強可以理解復(fù)雜任務(wù)并將其分解成清晰的步驟列表Step-by-step Plan。這個模型負責(zé)“想”即任務(wù)分解和步驟規(guī)劃。我們可以使用LM Studio或Ollama在本地運行它響應(yīng)快且隱私性好。執(zhí)行與對話模型Executor對于每一步的具體執(zhí)行比如根據(jù)大綱寫一段文字、寫一段Python代碼、或者回答用戶的具體問題可以視情況選擇。對質(zhì)量要求高的步驟可以調(diào)用DeepSeek-V3或GPT-3.5-Turbo的API對簡單、格式化的響應(yīng)可以用更小的本地模型如Phi-3-mini。這個模型負責(zé)“做”即執(zhí)行具體指令。摘要與工具調(diào)用模型Helper用于生成對話摘要、判斷用戶意圖是提問、指令還是中斷信號、以及決定是否需要調(diào)用外部工具如計算器、搜索引擎。這個角色對能力要求相對較低可以使用非常輕量的模型如Gemma 2B以降低整體延遲和資源消耗。為什么這么選將“規(guī)劃”和“執(zhí)行”分離符合人類的思考方式也使得系統(tǒng)更模塊化、更易調(diào)試。規(guī)劃模型一次生成整個計劃執(zhí)行模型則專注于當(dāng)前步驟互不干擾。同時混合使用本地模型和云端API在成本、性能和隱私之間取得了平衡。3.2 記憶系統(tǒng)向量數(shù)據(jù)庫與結(jié)構(gòu)化存儲向量數(shù)據(jù)庫長期記憶我選擇ChromaDB。它輕量、易用Python集成度極高非常適合原型和中小型項目。我們將每個任務(wù)線程的所有“記憶片段”用戶輸入、AI輸出、生成的文檔塊都通過text-embedding-3-small這類嵌入模型轉(zhuǎn)換成向量存入ChromaDB的對應(yīng)集合Collection中集合名可以用任務(wù)ID命名。結(jié)構(gòu)化存儲任務(wù)狀態(tài)使用SQLite足矣。我們需要一張tasks表核心字段包括task_id(主鍵)user_id(區(qū)分不同用戶)goal(任務(wù)目標(biāo))current_state(如 “planning”, “executing_step_3”, “paused”)plan(JSON格式存儲的任務(wù)分解步驟)checkpoint(JSON格式存儲的進度快照如上一步輸出)created_at,updated_at3.3 控制中樞Agent框架與流程編排雖然可以完全手寫控制邏輯但使用一個成熟的Agent框架能事半功倍。這里我推薦LangChain或LlamaIndex。LangChain優(yōu)勢在于其豐富的“鏈”Chain和“智能體”Agent抽象對于構(gòu)建復(fù)雜的、多步驟的、帶工具調(diào)用的工作流非常順手。它的ConversationBufferMemory、ConversationSummaryMemory等組件可以直接用于管理短期記憶。LlamaIndex優(yōu)勢在于數(shù)據(jù)連接和檢索RAG能力超強其“查詢引擎”和“聊天引擎”的概念與我們的“任務(wù)線程”模型很契合。它擅長管理復(fù)雜的文檔上下文對于需要深度檢索外部知識的任務(wù)場景更友好。在這個項目中我傾向于使用LangChain因為它對工作流Workflow和狀態(tài)管理的抽象更符合我們的“狀態(tài)機”思維。我們可以用LLMChain來構(gòu)建規(guī)劃器用AgentExecutor來運行每一步并用自定義的回調(diào)Callback函數(shù)來在特定節(jié)點如每一步結(jié)束后觸發(fā)狀態(tài)保存。4. 分步實現(xiàn)從任務(wù)創(chuàng)建到中斷恢復(fù)的全流程下面我們進入具體的代碼實現(xiàn)環(huán)節(jié)。我會用偽代碼和關(guān)鍵代碼片段來說明核心流程你可以根據(jù)自己的環(huán)境進行調(diào)整。4.1 第一步初始化系統(tǒng)與定義數(shù)據(jù)結(jié)構(gòu)首先定義我們的核心數(shù)據(jù)結(jié)構(gòu)。# 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] # 任務(wù)分解后的步驟列表 current_step_index: int # 當(dāng)前執(zhí)行到第幾步 checkpoint: Dict[str, Any] # 進度快照存任何需要的信息 created_at: datetime updated_at: datetime def to_dict(self): # 方便序列化存儲到數(shù)據(jù)庫 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]): # 從數(shù)據(jù)庫加載 data[status] TaskStatus(data[status]) data[created_at] datetime.fromisoformat(data[created_at]) data[updated_at] datetime.fromisoformat(data[updated_at]) return cls(**data)然后初始化我們的核心組件模型、記憶存儲和數(shù)據(jù)庫連接。# system_init.py import sqlite3 from langchain.llms import Ollama # 假設(shè)使用本地Ollama運行Llama3 from langchain.embeddings import HuggingFaceEmbeddings import chromadb # 1. 初始化規(guī)劃模型本地 planner_llm Ollama(modelllama3.1:8b, temperature0.1) # 低temperature保證規(guī)劃穩(wěn)定性 # 2. 初始化執(zhí)行模型這里示例用同一個實踐中可用不同的 executor_llm Ollama(modelllama3.1:8b, temperature0.7) # 高一點更有創(chuàng)造性 # 3. 初始化嵌入模型和向量數(shù)據(jù)庫長期記憶 embed_model HuggingFaceEmbeddings(model_nameall-MiniLM-L6-v2) # 輕量級嵌入模型 chroma_client chromadb.PersistentClient(path./chroma_db) # 注意我們不會創(chuàng)建一個全局的collection而是為每個任務(wù)動態(tài)創(chuàng)建/獲取 # 4. 初始化SQLite數(shù)據(jù)庫任務(wù)狀態(tài)存儲 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 第二步實現(xiàn)任務(wù)規(guī)劃與分解模塊這個模塊負責(zé)接收用戶模糊的目標(biāo)并將其轉(zhuǎn)化為清晰的步驟列表。# 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 你是一個高級任務(wù)規(guī)劃AI。請將用戶的目標(biāo)分解成一個清晰、有序、可執(zhí)行的步驟列表。 每個步驟應(yīng)該足夠具體以便另一個AI可以獨立執(zhí)行它。 考慮可能需要的迭代或用戶確認點。 目標(biāo){goal} 請以嚴格的JSON數(shù)組格式輸出每個元素是一個包含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規(guī)劃失敗: {e}) # 降級方案返回一個簡單的默認步驟 return [TaskStep(step_id1, descriptionf執(zhí)行任務(wù){(diào)goal}, statuspending)]4.3 第三步構(gòu)建任務(wù)執(zhí)行引擎與狀態(tài)保存這是最核心的部分它驅(qū)動任務(wù)一步步前進并在每一步之后保存狀態(tài)。# 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: 創(chuàng)建新任務(wù)并初始化規(guī)劃 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, # 還未開始執(zhí)行任何步驟 checkpoint{goal: goal, history: []}, created_atnow, updated_atnow ) # 保存到數(shù)據(jù)庫 self._save_task_to_db(task) # 為這個任務(wù)創(chuàng)建專屬的向量集合長期記憶 collection self.chroma_client.create_collection(namefmemory_{task_id}) # 初始記憶任務(wù)目標(biāo) collection.add( documents[f任務(wù)目標(biāo){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): 執(zhí)行當(dāng)前步驟返回執(zhí)行結(jié)果和是否完成標(biāo)志 task self._load_task_from_db(task_id) if task.status ! TaskStatus.EXECUTING and task.status ! TaskStatus.PLANNING: return f任務(wù)狀態(tài)為{task.status.value}無法執(zhí)行。, 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. 準(zhǔn)備上下文從checkpoint和長期記憶中獲取相關(guān)信息 context self._retrieve_relevant_memory(task_id, current_step.description) # 2. 構(gòu)建給執(zhí)行模型的提示 prompt f 你正在協(xié)助用戶完成一個任務(wù)。 總目標(biāo){task.goal} 當(dāng)前步驟第{current_step.step_id}步{current_step.description} 歷史上下文 {context} 請執(zhí)行當(dāng)前步驟。輸出應(yīng)清晰、完整直接給出步驟的結(jié)果。 輸出 # 3. 調(diào)用模型執(zhí)行 try: step_result self.llm.invoke(prompt) except Exception as e: step_result f步驟執(zhí)行出錯{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結(jié)果{step_result}], metadatas[{type: step_result, step: current_step.step_id}], ids[fstep_{current_step.step_id}_{task_id}] ) # 5. 更新任務(wù)狀態(tài)和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} 完成。任務(wù)全部完成\n結(jié)果{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結(jié)果{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 用戶請求中斷): 暫停任務(wù) 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任務(wù)于{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: 恢復(fù)被暫停的任務(wù) 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) # 生成恢復(fù)提示 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結(jié)果摘要{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接下來繼續(xù){next_step.description} if next_step else 所有步驟已完成。 resume_msg f任務(wù)已恢復(fù)。{last_step_info}{next_step_info} return resume_msg else: return f任務(wù)當(dāng)前狀態(tài)為{task.status.value}無法恢復(fù)。 def _retrieve_relevant_memory(self, task_id: str, query: str, n_results: int 3) - str: 從該任務(wù)的長期記憶中檢索相關(guān)信息 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 暫無相關(guān)歷史記憶 def _save_task_to_db(self, task: Task): # ... (實現(xiàn)數(shù)據(jù)庫插入邏輯) pass def _load_task_from_db(self, task_id: str) - Task: # ... (實現(xiàn)數(shù)據(jù)庫查詢邏輯) pass def _update_task_in_db(self, task: Task): # ... (實現(xiàn)數(shù)據(jù)庫更新邏輯) pass4.4 第四步設(shè)計主控循環(huán)與用戶交互接口最后我們需要一個主循環(huán)來連接用戶輸入和我們的引擎。# main_controller.py import re class AgentController: def __init__(self, engine: ExecutionEngine): self.engine engine self.active_tasks {} # user_id - task_id 的映射簡化處理實際應(yīng)存數(shù)據(jù)庫 def process_input(self, user_id: str, user_input: str) - str: # 1. 意圖識別是新建任務(wù)、控制指令還是繼續(xù)對話 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 # 開始執(zhí)行第一步 task.status TaskStatus.EXECUTING self.engine._update_task_in_db(task) result, finished self.engine.execute_step(task.task_id) return f已創(chuàng)建任務(wù)【{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 任務(wù)已暫停。你可以隨時輸入 /resume 來恢復(fù)。 else: return 當(dāng)前沒有正在執(zhí)行的任務(wù)可暫停。 return 你當(dāng)前沒有活躍任務(wù)。 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) # 恢復(fù)后自動執(zhí)行下一步 result, finished self.engine.execute_step(task_id) return f{resume_msg}\n\n{result} return 你當(dāng)前沒有可恢復(fù)的任務(wù)。請使用 /new 創(chuàng)建一個新任務(wù)。 elif user_input.lower().startswith(/list): # 查詢該用戶的所有任務(wù)簡化示例 return self._list_user_tasks(user_id) # 2. 如果是普通輸入且當(dāng)前有活躍任務(wù)則視為對當(dāng)前步驟的補充或反饋 elif user_id in self.active_tasks: task_id self.active_tasks[user_id] task self.engine._load_task_from_db(task_id) # 將用戶輸入作為額外信息存入記憶可能影響下一步執(zhí)行 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()}] ) # 然后繼續(xù)執(zhí)行下一步或者根據(jù)設(shè)計可以重新執(zhí)行當(dāng)前步 result, finished self.engine.execute_step(task_id) return f已記錄你的反饋。\n{result} # 3. 其他情況當(dāng)作新任務(wù)的開始簡化處理 else: return 似乎你想開始一個新任務(wù)請使用 /new [你的任務(wù)目標(biāo)] 的格式告訴我例如/new 幫我寫一封英文會議邀請函 def _list_user_tasks(self, user_id: str) - str: # 查詢數(shù)據(jù)庫并格式化輸出 # ... (實現(xiàn)數(shù)據(jù)庫查詢) return 你的任務(wù)列表\n1. [任務(wù)ID] 寫周報 (狀態(tài)進行中)\n2. [任務(wù)ID] 策劃活動 (狀態(tài)已暫停)5. 常見問題、調(diào)試技巧與優(yōu)化方向在實際搭建和測試過程中我遇到了不少坑也總結(jié)出一些讓Agent更“聰明”的技巧。5.1 模型不按格式輸出怎么辦這是使用開源模型最常見的問題。規(guī)劃步驟時我們要求模型輸出JSON但它可能返回一段自由文本。解決方案強化提示詞Prompt Engineering在提示詞中明確給出格式并使用“json ...”這樣的標(biāo)記來框定輸出范圍。示例中我們已經(jīng)這樣做了。后處理解析像代碼中那樣嘗試提取兩個“”之間的內(nèi)容或者尋找第一個“[”和最后一個“]”之間的內(nèi)容。可以寫一個健壯的解析函數(shù)嘗試json.loads()如果失敗就用正則表達式清理文本再試。使用輸出解析器Output ParserLangChain提供了PydanticOutputParser、StructuredOutputParser等工具能強制模型按指定格式輸出大大降低解析失敗率。這是更優(yōu)雅的解決方案。5.2 任務(wù)分解不合理或步驟太粗/太細模型可能把“寫報告”分解成“打開電腦”、“打開Word”、“開始寫”這樣無意義的步驟或者相反步驟過于籠統(tǒng)。解決方案提供示例Few-shot Prompting在給規(guī)劃模型的提示詞中包含1-2個高質(zhì)量的任務(wù)分解示例。例如“目標(biāo)組織一次線上技術(shù)分享會。輸出示例[{step_id:1, description:確定分享主題和核心聽眾}, {step_id:2, description:邀請2-3位潛在的分享嘉賓并協(xié)調(diào)時間}, ...]”迭代規(guī)劃不要一次性生成所有步驟。可以先讓模型生成一個高層大綱如1. 準(zhǔn)備階段 2. 執(zhí)行階段 3. 收尾階段然后對每個階段再分別進行細化分解。這更符合人類的規(guī)劃習(xí)慣。人工審核介入點在規(guī)劃提示詞中明確“在涉及關(guān)鍵決策如預(yù)算分配、技術(shù)選型或需要外部信息如查詢某產(chǎn)品價格的步驟后應(yīng)標(biāo)記為need_human_input。”這樣Agent會在這些節(jié)點主動暫停等待用戶輸入。5.3 長期記憶檢索不準(zhǔn)導(dǎo)致上下文混亂向量檢索并非總是精準(zhǔn)可能召回不相關(guān)的記憶片段干擾模型判斷。解決方案優(yōu)化記憶片段Chunking存入向量數(shù)據(jù)庫的文本不宜過長或過短。一個步驟的結(jié)果可以作為一個片段。對于長文檔可以按語義段落切割。好的分塊策略能極大提升檢索質(zhì)量。混合檢索Hybrid Search不要只依賴向量相似度搜索。可以結(jié)合關(guān)鍵詞搜索如BM25。ChromaDB等數(shù)據(jù)庫支持混合檢索。對于確切的名稱、日期、數(shù)字關(guān)鍵詞搜索往往更準(zhǔn)。重排序Re-ranking先召回較多的候選片段比如10個再用一個更小、更快的交叉編碼器Cross-Encoder模型對它們進行相關(guān)性重排序只取Top 3給模型。這能顯著提升精度但會增加計算開銷。記憶元數(shù)據(jù)過濾在檢索時利用我們存入的元數(shù)據(jù)如{type: step_result, step: 2}進行過濾。例如當(dāng)執(zhí)行第5步時我們可以優(yōu)先檢索step為4或5的記憶而不是所有記憶。5.4 如何評估Agent的表現(xiàn)除了人工測試可以建立一些自動化評估指標(biāo)任務(wù)完成率給定10個標(biāo)準(zhǔn)任務(wù)有多少個能從頭到尾無需人工干預(yù)除了設(shè)計好的介入點地走完流程中斷恢復(fù)成功率在任意步驟暫停后恢復(fù)指令能否正確載入上下文并繼續(xù)恢復(fù)后的下一步執(zhí)行是否正確步驟合理性人工評分請多人對分解出的步驟進行1-5分打分評估其邏輯性和可執(zhí)行性。幻覺率在需要事實性知識的步驟中如“查詢2023年全球智能手機出貨量”Agent是否編造了不存在的數(shù)據(jù)搭建這個AI Agent的過程更像是在設(shè)計一套人機交互的“協(xié)議”和“工作流”。技術(shù)本身模型、向量數(shù)據(jù)庫是工具而如何讓這些工具流暢地協(xié)作理解并適應(yīng)人類非線性、多線程的工作方式才是真正的挑戰(zhàn)。這個原型僅僅是一個起點你可以在此基礎(chǔ)上增加更復(fù)雜的特性比如多任務(wù)并行管理、工具調(diào)用讓Agent能自己上網(wǎng)搜索、操作軟件、甚至讓多個Agent之間進行協(xié)作。希望這個詳細的拆解能給你提供一個堅實的起點去創(chuàng)造真正懂你、能與你并肩工作的AI伙伴。