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