1. 項(xiàng)目概述從“筆記”到“工程實(shí)踐”最近在整理Harness Engineering相關(guān)的技術(shù)文檔時(shí)我發(fā)現(xiàn)關(guān)于Agent運(yùn)行時(shí)核心機(jī)制的討論尤其是循環(huán)、路由與上下文這三塊是很多開發(fā)者從“會用框架”到“理解框架”的關(guān)鍵門檻。網(wǎng)上能找到的資料要么過于零散只講某個(gè)API怎么調(diào)用要么過于理論堆砌一堆架構(gòu)圖卻不說清楚代碼到底怎么跑起來的。這讓我覺得有必要結(jié)合自己踩過的坑把這塊“硬骨頭”啃下來寫一篇能直接指導(dǎo)編碼的深度解析。所謂Harness Engineering你可以把它理解為一套用于構(gòu)建、管理和控制智能體Agent的工程化框架與最佳實(shí)踐。它關(guān)注的不是某個(gè)單一的算法模型而是如何讓多個(gè)Agent協(xié)同、穩(wěn)定、高效地完成復(fù)雜任務(wù)。在這個(gè)體系里Agent運(yùn)行時(shí)就是那個(gè)讓智能體“活”起來、能夠感知、決策并執(zhí)行的核心引擎。而驅(qū)動這個(gè)引擎的三大核心機(jī)制正是循環(huán)Loop、路由Routing和上下文Context。理解它們你才能真的駕馭Agent而不是被框架牽著鼻子走。這篇文章適合誰呢如果你正在或打算開發(fā)基于Agent的應(yīng)用無論是自動化工作流、智能客服還是復(fù)雜的決策系統(tǒng)并且已經(jīng)過了“Hello World”階段開始頭疼于Agent的狀態(tài)管理、任務(wù)調(diào)度和信息流轉(zhuǎn)問題那么這里面的內(nèi)容就是為你準(zhǔn)備的。我會盡量避開空泛的概念用具體的代碼段、設(shè)計(jì)抉擇背后的“為什么”以及我實(shí)際調(diào)試中總結(jié)出的“坑點(diǎn)”來展開。2. 核心機(jī)制深度拆解循環(huán)、路由與上下文的角色與聯(lián)動在深入細(xì)節(jié)之前我們必須建立一個(gè)頂層的認(rèn)知模型這三個(gè)機(jī)制不是孤立的它們共同構(gòu)成了Agent運(yùn)行時(shí)的一個(gè)完整“心跳周期”。想象一下一個(gè)處理用戶咨詢的客服Agent。它拿到用戶問題上下文需要決定這個(gè)問題屬于售前、售后還是技術(shù)故障路由然后根據(jù)決定調(diào)用相應(yīng)的子能力或工具去執(zhí)行執(zhí)行后產(chǎn)生新的結(jié)果并評估是否還需要進(jìn)一步追問或可以結(jié)束循環(huán)。這個(gè)過程周而復(fù)始。2.1 循環(huán)LoopAgent執(zhí)行的狀態(tài)機(jī)與驅(qū)動力循環(huán)機(jī)制是Agent運(yùn)行的“節(jié)拍器”。它決定了Agent在單次觸發(fā)后如何推進(jìn)其內(nèi)部狀態(tài)何時(shí)停止。最簡單的循環(huán)就是“執(zhí)行一次就結(jié)束”O(jiān)ne-Shot。但對于復(fù)雜任務(wù)我們需要更強(qiáng)大的循環(huán)模式。2.1.1 常見循環(huán)模式與實(shí)踐選擇While-Loop條件循環(huán) 這是最常用、最靈活的循環(huán)模式。它基于一個(gè)或多個(gè)條件來決定是否繼續(xù)迭代。例如一個(gè)數(shù)據(jù)分析Agent可能會循環(huán)執(zhí)行“查詢-分析-提煉”步驟直到分析結(jié)果的置信度達(dá)到某個(gè)閾值或者用戶明確要求停止。# 偽代碼示例基于條件的While循環(huán) context initialize_context(user_query) while not should_stop(context): # 1. 路由決策根據(jù)當(dāng)前context決定下一步動作 action routing_engine.decide(context) # 2. 執(zhí)行動作 result execute_action(action, context) # 3. 更新上下文 context.update(result, action) # 4. 評估停止條件例如任務(wù)完成、達(dá)到最大步數(shù)、用戶中斷 context.evaluate_stop_condition()實(shí)操心得should_stop函數(shù)的設(shè)計(jì)是核心。不要只依賴單一條件如任務(wù)標(biāo)記完成最好結(jié)合多個(gè)維度最大迭代次數(shù)防止死循環(huán)、用戶主動取消信號、連續(xù)多次迭代未產(chǎn)生有效進(jìn)展等。我通常會設(shè)置一個(gè)默認(rèn)的最大步數(shù)比如50步作為安全網(wǎng)。For-Loop有限循環(huán) 適用于步驟明確、次數(shù)固定的任務(wù)。比如一個(gè)Agent需要依次檢查系統(tǒng)的A、B、C三個(gè)服務(wù)狀態(tài)。checklist [檢查數(shù)據(jù)庫連接, 驗(yàn)證API網(wǎng)關(guān), 掃描日志錯(cuò)誤] for task in checklist: context.set_current_task(task) result execute_standard_check(task, context) context.record_check_result(task, result)注意事項(xiàng)即使在For-Loop中也要為每個(gè)步驟設(shè)計(jì)異常處理。某個(gè)步驟的失敗不應(yīng)導(dǎo)致整個(gè)Agent崩潰而應(yīng)該被捕獲并記錄到上下文中供后續(xù)步驟或最終匯總時(shí)參考。遞歸循環(huán)Recursive Loop 當(dāng)任務(wù)可以分解為同構(gòu)的子任務(wù)時(shí)使用。例如一個(gè)文件整理Agent遇到文件夾就遞歸處理其中的文件。核心挑戰(zhàn)上下文的管理。遞歸每一層都需要有自己的上下文“快照”同時(shí)又能訪問到必要的父級信息。需要清晰界定上下文的作用域和繼承關(guān)系否則很容易造成信息污染或丟失。2.1.2 循環(huán)控制的高級技巧暫停與恢復(fù)Pause/Resume對于長時(shí)任務(wù)Agent需要支持暫停。關(guān)鍵在于將完整的運(yùn)行時(shí)狀態(tài)包括循環(huán)索引、當(dāng)前上下文、中間結(jié)果序列化并持久化?；謴?fù)時(shí)再反序列化從斷點(diǎn)繼續(xù)。這通常需要框架層面的支持。并行循環(huán)當(dāng)多個(gè)子任務(wù)相互獨(dú)立時(shí)可以使用并行循環(huán)來提升效率。但要注意共享上下文資源的線程安全問題。一個(gè)穩(wěn)妥的做法是采用“復(fù)制-合并”策略每個(gè)并行任務(wù)處理上下文的一個(gè)副本執(zhí)行完畢后再將結(jié)果合并回主上下文。循環(huán)超時(shí)與看門狗Watchdog必須為每個(gè)循環(huán)設(shè)置超時(shí)機(jī)制。一個(gè)獨(dú)立的看門狗線程可以監(jiān)控主循環(huán)的執(zhí)行時(shí)間一旦超時(shí)就觸發(fā)中斷流程保存現(xiàn)場并上報(bào)錯(cuò)誤避免Agent“卡死”。2.2 路由RoutingAgent的決策中樞與流量控制器如果說循環(huán)是“節(jié)奏”那么路由就是“方向”。它負(fù)責(zé)在運(yùn)行時(shí)根據(jù)當(dāng)前上下文動態(tài)決定下一步該執(zhí)行哪個(gè)動作、調(diào)用哪個(gè)工具、或者將任務(wù)移交給哪個(gè)更專業(yè)的子Agent。2.2.1 路由策略解析路由的核心是一個(gè)決策函數(shù)Input(Current_Context) - Output(Next_Action/Agent)?；谝?guī)則的路由Rule-Based 最簡單直接的方式。通過預(yù)定義的if-else或決策樹來路由。def rule_based_router(context): if 退款 in context.user_intent: return refund_agent elif 技術(shù)故障 in context.sentiment and context.user_tier VIP: return premium_support_agent else: return general_support_agent優(yōu)點(diǎn)確定性強(qiáng)易于調(diào)試和解釋。缺點(diǎn)規(guī)則膨脹后難以維護(hù)靈活性差無法處理未預(yù)見的情況?；谀Ｐ偷穆酚蒑odel-Based 利用機(jī)器學(xué)習(xí)模型如分類器、甚至小型LLM來學(xué)習(xí)從上下文到最佳動作的映射。這是當(dāng)前復(fù)雜Agent系統(tǒng)的趨勢。# 使用嵌入Embedding和向量相似度進(jìn)行路由 from sentence_transformers import SentenceTransformer import numpy as np class EmbeddingRouter: def __init__(self): self.model SentenceTransformer(all-MiniLM-L6-v2) # 預(yù)定義動作及其描述 self.action_descriptions { search_db: 在知識庫中搜索相關(guān)信息, call_api: 調(diào)用外部API獲取實(shí)時(shí)數(shù)據(jù), generate_report: 生成分析總結(jié)報(bào)告 } # 預(yù)計(jì)算動作描述的向量 self.action_vectors {name: self.model.encode(desc) for name, desc in self.action_descriptions.items()} def route(self, context): # 將當(dāng)前上下文如最新用戶問題編碼為向量 query_vector self.model.encode(context.latest_query) # 計(jì)算與所有動作的余弦相似度 similarities {} for action_name, action_vec in self.action_vectors.items(): cos_sim np.dot(query_vector, action_vec) / (np.linalg.norm(query_vector) * np.linalg.norm(action_vec)) similarities[action_name] cos_sim # 返回最相似的動作 return max(similarities, keysimilarities.get)實(shí)操心得模型路由的關(guān)鍵在于高質(zhì)量的動作描述和上下文特征工程。動作描述要精準(zhǔn)概括其功能和適用場景。上下文不能只扔進(jìn)模型需要提煉出關(guān)鍵特征如用戶意圖、對話歷史摘要、當(dāng)前任務(wù)階段等。混合路由Hybrid 結(jié)合規(guī)則和模型的優(yōu)勢。通常用規(guī)則處理明確、高優(yōu)先級的場景如安全攔截、緊急轉(zhuǎn)人工用模型處理復(fù)雜的、模糊的決策。也可以先用模型給出幾個(gè)候選再用規(guī)則進(jìn)行篩選和排序。2.2.2 路由表與策略鏈在工程實(shí)現(xiàn)上路由決策往往不是一步完成的。路由表Routing Table一個(gè)可配置的映射表將狀態(tài)或意圖映射到處理單元。適合規(guī)則路由便于運(yùn)營人員修改。策略鏈Policy Chain按順序執(zhí)行多個(gè)路由策略。例如先經(jīng)過一個(gè)“過濾器策略”排除非法請求再經(jīng)過一個(gè)“優(yōu)先級策略”識別VIP用戶最后經(jīng)過一個(gè)“負(fù)載均衡策略”分配Agent實(shí)例。每個(gè)策略都可以對路由結(jié)果進(jìn)行修改或傳遞。注意路由決策本身也是有成本的。要避免在每次循環(huán)中都進(jìn)行非常復(fù)雜的模型推理。可以考慮對路由結(jié)果進(jìn)行緩存例如相同上下文指紋在短時(shí)間內(nèi)路由到相同目標(biāo)或者將路由決策的粒度放粗每N步或任務(wù)階段變更時(shí)才重新路由。2.3 上下文ContextAgent的“記憶”與“工作臺”上下文是貫穿循環(huán)和路由的“血液”。它承載了Agent任務(wù)執(zhí)行過程中的所有狀態(tài)和信息。一個(gè)設(shè)計(jì)良好的上下文管理機(jī)制是構(gòu)建穩(wěn)定、可維護(hù)Agent系統(tǒng)的基石。2.3.1 上下文的層次結(jié)構(gòu)與內(nèi)容上下文不是一個(gè)大雜燴字典而應(yīng)該有清晰的結(jié)構(gòu)會話上下文Session Context范圍一次用戶會話可能包含多輪對話的全局信息。內(nèi)容用戶ID、會話ID、創(chuàng)建時(shí)間、全局配置、用戶長期偏好、安全令牌等。生命周期從會話開始到結(jié)束。任務(wù)上下文Task Context范圍一個(gè)具體任務(wù)可能跨多個(gè)循環(huán)的信息。內(nèi)容任務(wù)目標(biāo)、任務(wù)參數(shù)、當(dāng)前任務(wù)狀態(tài)如進(jìn)行中、暫停、完成、任務(wù)歷史步驟記錄、中間結(jié)果。生命周期從任務(wù)創(chuàng)建到任務(wù)終結(jié)成功、失敗或取消?；睾仙舷挛腡urn Context范圍單次循環(huán)或單次用戶交互的信息。內(nèi)容本輪的用戶輸入、上輪Agent的輸出、當(dāng)前路由決策、本次執(zhí)行的動作及其結(jié)果、臨時(shí)變量。生命周期一次循環(huán)開始到結(jié)束。2.3.2 上下文的數(shù)據(jù)流與持久化上下文數(shù)據(jù)需要在循環(huán)、路由以及不同的處理單元間流動。數(shù)據(jù)流設(shè)計(jì)建議采用不可變Immutable或?qū)憰r(shí)復(fù)制Copy-on-Write的理念。每次循環(huán)或動作執(zhí)行產(chǎn)生的新結(jié)果應(yīng)生成一個(gè)新的上下文版本或更新一個(gè)特定的分支而不是直接修改全局上下文。這有利于調(diào)試、回滾和實(shí)現(xiàn)“撤銷”功能。class TurnContext: def __init__(self, previous_context, new_input): self.session_id previous_context.session_id # 繼承 self.task_state previous_context.task_state.copy() # 淺拷貝或深拷貝視情況而定 self.current_input new_input self.actions_executed [] # 本輪新增 def record_action(self, action, result): self.actions_executed.append({action: action, result: result}) # 根據(jù)結(jié)果更新任務(wù)狀態(tài) self.task_state.update_from_result(result)持久化策略何時(shí)持久化關(guān)鍵節(jié)點(diǎn)必須持久化如任務(wù)開始、任務(wù)完成、每輪循環(huán)結(jié)束尤其是長循環(huán)、以及發(fā)生錯(cuò)誤時(shí)。存儲什么至少需要存儲足以恢復(fù)任務(wù)狀態(tài)的最小數(shù)據(jù)集。包括上下文的核心字段、循環(huán)的進(jìn)度標(biāo)識、路由決策歷史等。存儲后端根據(jù)延遲和一致性要求選擇。Redis適合做高速緩存PostgreSQL/MongoDB適合做可靠存儲。復(fù)雜的上下文對象可能需要序列化如JSON、MessagePack、Pickle后存儲。實(shí)操心得為上下文對象設(shè)計(jì)一個(gè)版本號version字段。當(dāng)你的Agent代碼升級上下文結(jié)構(gòu)可能變化版本號可以幫助你進(jìn)行數(shù)據(jù)遷移或兼容性處理。2.3.3 上下文壓縮與摘要隨著對話或任務(wù)進(jìn)行上下文會不斷膨脹尤其是包含長對話歷史或大量中間數(shù)據(jù)這會導(dǎo)致幾個(gè)問題1超出模型token限制2降低路由和決策效率3增加存儲和傳輸開銷。因此上下文壓縮Context Compression是高級Agent系統(tǒng)的必備技能?；瑒哟翱谥槐Ａ糇罱麼輪對話。最簡單但可能丟失關(guān)鍵早期信息。關(guān)鍵信息提取使用一個(gè)輕量級模型或規(guī)則從歷史上下文中提取出與當(dāng)前任務(wù)最相關(guān)的實(shí)體、意圖、事實(shí)結(jié)論丟棄冗余細(xì)節(jié)。增量式摘要在每輪或每N輪后動態(tài)生成一個(gè)不斷更新的對話摘要。新的決策基于這個(gè)摘要和最新輸入而不是全部原始?xì)v史。# 偽代碼簡單的增量摘要 class ConversationSummarizer: def __init__(self): self.summary def update_summary(self, new_dialogue_turn): # 將當(dāng)前摘要和新對話輪次一起讓LLM生成新的摘要 prompt f 現(xiàn)有摘要{self.summary} 最新一輪對話 用戶{new_dialogue_turn.user} Agent{new_dialogue_turn.agent} 請基于以上信息更新對話摘要保留所有關(guān)鍵決策、事實(shí)和待辦事項(xiàng)。 更新后的摘要 self.summary llm_invoke(prompt) return self.summary注意事項(xiàng)摘要的生成本身需要成本且可能存在信息損失。需要權(quán)衡摘要的更新頻率和精度。對于關(guān)鍵業(yè)務(wù)信息即使被摘要了也應(yīng)將其結(jié)構(gòu)化后單獨(dú)存儲在任務(wù)上下文中。3. 實(shí)戰(zhàn)構(gòu)建一個(gè)具備核心運(yùn)行機(jī)制的簡易任務(wù)處理Agent理論說再多不如動手搭一個(gè)。我們來設(shè)計(jì)一個(gè)簡易的“技術(shù)支持工單處理Agent”它需要理解用戶問題自動嘗試一些修復(fù)步驟如果不行則路由給人工客服。3.1 系統(tǒng)架構(gòu)與組件設(shè)計(jì)我們將系統(tǒng)分為以下模塊主循環(huán)引擎MainLoopEngine控制整體執(zhí)行流程。上下文管理器ContextManager創(chuàng)建、更新、持久化上下文。路由決策器Router基于上下文決定下一步。動作執(zhí)行器ActionExecutor執(zhí)行具體的工具調(diào)用或子Agent任務(wù)。持久化存儲Storage用于保存上下文和會話狀態(tài)。3.2 核心代碼實(shí)現(xiàn)解析3.2.1 上下文定義from dataclasses import dataclass, asdict, field from typing import Dict, List, Any, Optional from datetime import datetime import json dataclass class SessionContext: 會話級上下文 session_id: str user_id: str created_at: datetime attributes: Dict[str, Any] field(default_factorydict) # 存放用戶偏好等 dataclass class TaskContext: 任務(wù)級上下文一個(gè)工單 task_id: str session_id: str description: str # 用戶問題描述 status: str open # open, in_progress, resolved, escalated created_at: datetime field(default_factorydatetime.now) updated_at: datetime field(default_factorydatetime.now) max_steps: int 10 # 最大自動處理步數(shù) current_step: int 0 history: List[Dict] field(default_factorylist) # 記錄每一步操作 extracted_info: Dict[str, Any] field(default_factorydict) # 提取的關(guān)鍵信息如錯(cuò)誤代碼、設(shè)備型號 dataclass class TurnContext: 回合級上下文 turn_id: int task_context: TaskContext user_input: str router_decision: Optional[str] None action_result: Optional[Dict] None error: Optional[str] None def to_dict(self): # 方便序列化 return asdict(self)3.2.2 主循環(huán)引擎實(shí)現(xiàn)class MainLoopEngine: def __init__(self, router, action_executor, storage): self.router router self.action_executor action_executor self.storage storage def run_task(self, session_ctx: SessionContext, initial_query: str) - TaskContext: # 1. 初始化任務(wù)上下文 task_ctx TaskContext( task_idftask_{datetime.now().timestamp()}, session_idsession_ctx.session_id, descriptioninitial_query ) self.storage.save_task(task_ctx) # 2. 主循環(huán) while task_ctx.status in [open, in_progress]: # 檢查最大步數(shù)限制 if task_ctx.current_step task_ctx.max_steps: task_ctx.status escalated task_ctx.history.append({step: task_ctx.current_step, action: max_steps_reached, result: 自動處理步數(shù)超限轉(zhuǎn)人工}) break # 創(chuàng)建本輪上下文 turn_ctx TurnContext( turn_idtask_ctx.current_step, task_contexttask_ctx, user_inputinitial_query if task_ctx.current_step 0 else # 后續(xù)輪次可能無新輸入 ) # 3. 路由決策 try: next_action self.router.decide(turn_ctx) turn_ctx.router_decision next_action except Exception as e: turn_ctx.error f路由決策失敗: {e} task_ctx.status escalated self._record_turn(task_ctx, turn_ctx) break # 4. 執(zhí)行動作 try: result self.action_executor.execute(next_action, turn_ctx) turn_ctx.action_result result # 根據(jù)結(jié)果更新任務(wù)狀態(tài) self._update_task_from_result(task_ctx, result) except Exception as e: turn_ctx.error f動作執(zhí)行失敗: {e} # 執(zhí)行失敗可以考慮重試或直接升級 task_ctx.status escalated self._record_turn(task_ctx, turn_ctx) break # 5. 記錄本輪并更新循環(huán)狀態(tài) self._record_turn(task_ctx, turn_ctx) task_ctx.current_step 1 task_ctx.updated_at datetime.now() # 6. 持久化任務(wù)狀態(tài)檢查點(diǎn) if task_ctx.current_step % 3 0: # 每3步持久化一次 self.storage.save_task(task_ctx) # 循環(huán)結(jié)束最終持久化 self.storage.save_task(task_ctx) return task_ctx def _record_turn(self, task_ctx: TaskContext, turn_ctx: TurnContext): 記錄單輪執(zhí)行歷史 record { step: turn_ctx.turn_id, decision: turn_ctx.router_decision, action_result: turn_ctx.action_result, error: turn_ctx.error, timestamp: datetime.now().isoformat() } task_ctx.history.append(record) # 也可以選擇將詳細(xì)的turn_ctx單獨(dú)存儲 self.storage.save_turn(turn_ctx) def _update_task_from_result(self, task_ctx: TaskContext, result: Dict): 根據(jù)動作執(zhí)行結(jié)果更新任務(wù)狀態(tài) if result.get(status) resolved: task_ctx.status resolved elif result.get(suggest_escalation): task_ctx.status escalated # 更新提取的信息 if extracted_info in result: task_ctx.extracted_info.update(result[extracted_info])3.2.3 一個(gè)混合路由器的示例class HybridRouter: def __init__(self, rule_router, model_router, fallback_actionescalate_to_human): self.rule_router rule_router self.model_router model_router self.fallback_action fallback_action def decide(self, turn_ctx: TurnContext) - str: # 第一層安全與優(yōu)先級規(guī)則硬性規(guī)則 if self._is_urgent_issue(turn_ctx): return immediate_human_escalation if self._contains_sensitive_info(turn_ctx): return mask_and_process # 第二層基于規(guī)則的快速路由明確場景 rule_based_action self.rule_router.decide(turn_ctx) if rule_based_action and rule_based_action ! unknown: return rule_based_action # 第三層基于模型的智能路由模糊場景 model_based_action self.model_router.decide(turn_ctx) if model_based_action: return model_based_action # 默認(rèn)降級方案 return self.fallback_action def _is_urgent_issue(self, turn_ctx): # 判斷是否為緊急問題例如包含“宕機(jī)”、“無法使用”等關(guān)鍵詞且來自VIP用戶 urgent_keywords [宕機(jī), 崩潰, 完全無法] task_desc turn_ctx.task_context.description # 這里假設(shè)能從session_ctx獲取用戶等級簡化處理 return any(kw in task_desc for kw in urgent_keywords) def _contains_sensitive_info(self, turn_ctx): # 簡單示例檢測是否包含疑似密碼的信息 import re potential_pwd_pattern r(?i)(password|pwd|密碼)[:]\s*\S return bool(re.search(potential_pwd_pattern, turn_ctx.user_input or turn_ctx.task_context.description))3.3 動作執(zhí)行器的設(shè)計(jì)模式動作執(zhí)行器負(fù)責(zé)將路由決策的“動作名稱”轉(zhuǎn)化為具體的操作。一個(gè)好的模式是使用“注冊表”Registry。class ActionExecutor: def __init__(self): self._actions {} def register(self, name: str, action_func): self._actions[name] action_func def execute(self, action_name: str, turn_ctx: TurnContext) - Dict: if action_name not in self._actions: raise ValueError(f未知動作: {action_name}) try: # 執(zhí)行動作并傳入當(dāng)前上下文 result self._actions[action_name](turn_ctx) # 確保返回結(jié)果包含標(biāo)準(zhǔn)字段 result.setdefault(status, success) return result except Exception as e: # 記錄詳細(xì)的執(zhí)行錯(cuò)誤 return {status: error, message: str(e), action: action_name} # 使用示例 executor ActionExecutor() executor.register(search_knowledge_base) def search_kb_action(turn_ctx): query turn_ctx.task_context.description # 模擬搜索知識庫 search_results simulate_kb_search(query) # 嘗試從結(jié)果中提取解決方案 solution extract_solution(search_results) return { status: resolved if solution else no_match, data: search_results, proposed_solution: solution, extracted_info: {query_topic: query[:50]} # 記錄提取的信息 } executor.register(run_diagnostic_script) def run_diagnostic_action(turn_ctx): # 模擬運(yùn)行一個(gè)診斷腳本 diagnostic_result run_script(standard_diagnostic) if diagnostic_result.get(issues_found): return {status: in_progress, fix_steps: diagnostic_result[suggested_fixes]} else: return {status: no_issue_detected, suggest_escalation: True}4. 生產(chǎn)環(huán)境下的挑戰(zhàn)與調(diào)優(yōu)實(shí)錄把Demo跑起來只是第一步真正上線后各種意想不到的問題才會浮現(xiàn)。下面分享幾個(gè)我實(shí)踐中遇到的典型挑戰(zhàn)和解決思路。4.1 循環(huán)失控與死鎖預(yù)防問題場景一個(gè)用于處理文檔的Agent在“解析-提取-驗(yàn)證”的循環(huán)中因?yàn)槟硞€(gè)邊緣文檔格式解析異常導(dǎo)致驗(yàn)證永遠(yuǎn)無法通過循環(huán)卡死直到達(dá)到最大步數(shù)才超時(shí)退出浪費(fèi)資源且體驗(yàn)差。根因分析循環(huán)的停止條件過于依賴業(yè)務(wù)結(jié)果如“驗(yàn)證通過”未考慮異常狀態(tài)。缺乏對循環(huán)內(nèi)“無進(jìn)展”狀態(tài)的檢測。解決方案設(shè)置多維停止條件除了業(yè)務(wù)成功狀態(tài)和最大步數(shù)增加“無進(jìn)展檢測”。def should_stop(task_ctx, turn_history): # 條件1: 業(yè)務(wù)成功 if task_ctx.status resolved: return True # 條件2: 達(dá)到最大步數(shù) if task_ctx.current_step task_ctx.max_steps: return True # 條件3: 檢測無進(jìn)展循環(huán) (最近N步結(jié)果高度相似或錯(cuò)誤相同) recent_steps turn_history[-5:] if len(turn_history) 5 else turn_history if len(recent_steps) 3: last_three_actions [step.get(decision) for step in recent_steps[-3:]] last_three_errors [step.get(error) for step in recent_steps[-3:]] # 如果連續(xù)三步路由到同一個(gè)無效動作或產(chǎn)生相同錯(cuò)誤 if (len(set(last_three_actions)) 1 and last_three_actions[0] in [action_a, action_b]) or \ (len(set(last_three_errors)) 1 and last_three_errors[0] is not None): return True # 觸發(fā)停止并標(biāo)記為需人工干預(yù) return False引入看門狗Watchdog線程在主循環(huán)外啟動一個(gè)監(jiān)控線程檢查單次循環(huán)的執(zhí)行時(shí)間。如果某次循環(huán)執(zhí)行時(shí)間遠(yuǎn)超歷史平均時(shí)間例如3個(gè)標(biāo)準(zhǔn)差以外則向主線程發(fā)送中斷信號保存當(dāng)前狀態(tài)并標(biāo)記為“超時(shí)異?！薄?.2 路由決策的準(zhǔn)確性與效率平衡問題場景使用一個(gè)大型語言模型LLM作為路由決策器雖然準(zhǔn)確率高但每個(gè)請求都調(diào)用LLM導(dǎo)致響應(yīng)延遲高、成本昂貴。根因分析路由決策的粒度太細(xì)且未利用緩存。解決方案分層路由與緩存第一層意圖快速分類。使用一個(gè)輕量級文本分類模型如FastText、小規(guī)模BERT或關(guān)鍵詞匹配將請求分到幾個(gè)大的意圖桶如“查詢”、“操作”、“故障”。這一步速度極快可以覆蓋80%的常見請求。第二層桶內(nèi)精細(xì)路由。只有進(jìn)入“故障”等復(fù)雜桶的請求才觸發(fā)更精細(xì)的LLM路由或規(guī)則路由。同時(shí)對路由結(jié)果進(jìn)行緩存。緩存的Key可以是“意圖桶用戶輸入文本的哈希”或“上下文特征指紋”。設(shè)置合理的TTL如5分鐘在短時(shí)間內(nèi)相同問題無需重復(fù)計(jì)算路由。路由決策預(yù)熱與異步更新對于已知的、高頻的任務(wù)模板可以在系統(tǒng)啟動時(shí)或低峰期預(yù)計(jì)算其路由結(jié)果存入緩存。對于模型路由可以異步定期更新模型而不影響在線推理路徑。4.3 上下文膨脹與信息丟失的權(quán)衡問題場景一個(gè)多輪對話Agent隨著對話輪次增加上下文越來越長最終超出模型Token限制。使用簡單的滑動窗口又丟失了對話開頭約定的關(guān)鍵約束條件。根因分析上下文管理策略過于簡單對所有信息一視同仁。解決方案結(jié)構(gòu)化上下文與重要性標(biāo)注不要將所有歷史都作為非結(jié)構(gòu)化的文本堆砌。設(shè)計(jì)結(jié)構(gòu)化的上下文對象將信息分類系統(tǒng)指令高優(yōu)先級Agent的角色、核心約束、對話目標(biāo)。這部分應(yīng)始終保留。關(guān)鍵事實(shí)與參數(shù)中優(yōu)先級用戶明確提供的實(shí)體、數(shù)字、選擇等?？梢蕴崛〕鰜矸旁谝粋€(gè)“事實(shí)表”中。對話歷史低優(yōu)先級具體的每一輪問答。這部分是壓縮的主要對象。動態(tài)摘要與關(guān)鍵信息提取在每輪對話后不是保存全部原文而是用一個(gè)小模型或提示詞工程生成一個(gè)增量摘要。這個(gè)摘要會融合上一輪的摘要和本輪的新內(nèi)容。同時(shí)運(yùn)行一個(gè)命名實(shí)體識別NER或信息提取流程將本輪出現(xiàn)的新關(guān)鍵信息如訂單號、日期、產(chǎn)品型號提取出來添加到“事實(shí)表”中。后續(xù)的路由和動作執(zhí)行主要參考“系統(tǒng)指令”、“事實(shí)表”和“最新摘要”而非全部歷史。只有當(dāng)需要深度理解某段歷史時(shí)才去查詢詳細(xì)的、經(jīng)過索引的原始記錄如果保存了的話。class CompressingContextManager: def __init__(self, max_raw_turns10): self.system_instructions self.fact_table {} # 鍵值對存儲關(guān)鍵事實(shí) self.dialogue_summary self.raw_turns [] # 保留最近N輪原始記錄用于追溯 self.max_raw_turns max_raw_turns def add_turn(self, user_input, agent_response): # 1. 保存原始記錄滑動窗口 self.raw_turns.append((user_input, agent_response)) if len(self.raw_turns) self.max_raw_turns: self.raw_turns.pop(0) # 2. 提取關(guān)鍵事實(shí) new_facts extract_facts(user_input, agent_response) self.fact_table.update(new_facts) # 3. 更新摘要 update_prompt f當(dāng)前摘要{self.dialogue_summary}\n新對話用戶說{user_input}Agent回復(fù){agent_response}\n請生成更新的摘要。 self.dialogue_summary call_llm_for_summary(update_prompt) def get_compressed_context_for_agent(self): 提供給Agent模型使用的壓縮后上下文 return { system: self.system_instructions, facts: json.dumps(self.fact_table, ensure_asciiFalse), summary: self.dialogue_summary, recent_raw: self.raw_turns[-2:] if self.raw_turns else [] # 提供最近一兩輪原文供參考 }4.4 調(diào)試與可觀測性建設(shè)當(dāng)Agent行為不符合預(yù)期時(shí)如何快速定位是循環(huán)、路由還是上下文的問題結(jié)構(gòu)化日志不要在代碼里隨意打印print。為每個(gè)循環(huán)步驟、路由決策、上下文更新、動作執(zhí)行記錄結(jié)構(gòu)化的日志。日志應(yīng)包含時(shí)間戳、會話ID、任務(wù)ID、步驟ID、組件名如Router、ActionX、日志級別、關(guān)鍵數(shù)據(jù)如路由決策結(jié)果、上下文快照的哈希、動作執(zhí)行耗時(shí)和錯(cuò)誤信息。# 使用結(jié)構(gòu)化日志庫如structlog或json logger import structlog logger structlog.get_logger() def some_router_function(context): log logger.bind(task_idcontext.task_id, turncontext.current_step) log.info(router.started, input_snippetcontext.user_input[:100]) try: decision make_decision(context) log.info(router.decision_made, decisiondecision, confidence0.85) return decision except Exception as e: log.error(router.failed, errorstr(e), context_snapshotcontext.to_dict()) raise分布式追蹤在微服務(wù)或分布式Agent架構(gòu)中使用OpenTelemetry等工具為每個(gè)用戶請求注入追蹤ID。這個(gè)ID貫穿整個(gè)調(diào)用鏈從接收請求到循環(huán)的每一步到路由到子服務(wù)調(diào)用讓你能在儀表盤上清晰地看到一個(gè)請求的完整生命周期和耗時(shí)瓶頸。上下文快照與回放將每個(gè)關(guān)鍵步驟尤其是路由決策前后和動作執(zhí)行前后的完整上下文對象序列化后存儲到可查詢的存儲中如Elasticsearch。當(dāng)出現(xiàn)問題時(shí)你可以通過任務(wù)ID檢索出完整的上下文歷史在本地或測試環(huán)境“回放”該任務(wù)精確復(fù)現(xiàn)問題。路由決策的可解釋性對于基于模型的路由不僅要輸出決策還要輸出置信度分?jǐn)?shù)和如果可能主要依據(jù)的特征。這能幫助你在調(diào)試時(shí)判斷是模型不準(zhǔn)還是輸入特征有問題。5. 性能優(yōu)化與擴(kuò)展性考量當(dāng)你的Agent系統(tǒng)從原型走向生產(chǎn)服務(wù)大量并發(fā)用戶時(shí)性能和擴(kuò)展性就成為必須面對的問題。5.1 運(yùn)行時(shí)性能優(yōu)化循環(huán)異步化如果Agent的某些動作是I/O密集型的如調(diào)用外部API、查詢數(shù)據(jù)庫不要讓主循環(huán)同步等待?？梢詫幼魈峤坏饺蝿?wù)隊(duì)列如Celery、RabbitMQ主循環(huán)只負(fù)責(zé)生成任務(wù)和監(jiān)聽結(jié)果。這樣單個(gè)Agent實(shí)例可以同時(shí)管理多個(gè)任務(wù)的執(zhí)行流極大提高吞吐量。路由決策批處理對于請求量大的場景可以將短時(shí)間內(nèi)的一批路由請求收集起來批量發(fā)送給模型進(jìn)行推理如果模型支持批量推理這比逐個(gè)請求效率高得多。上下文緩存會話級和任務(wù)級的上下文是高頻訪問對象。使用內(nèi)存緩存如Redis存儲活躍的上下文對象避免每次循環(huán)都從數(shù)據(jù)庫讀取。注意設(shè)計(jì)合理的緩存失效和回寫策略。動作執(zhí)行結(jié)果緩存對于一些冪等的、結(jié)果相對穩(wěn)定的動作如“根據(jù)城市名查詢天氣”可以對其結(jié)果進(jìn)行緩存。路由決策后先查緩存命中則直接返回避免重復(fù)執(zhí)行。5.2 系統(tǒng)擴(kuò)展性設(shè)計(jì)無狀態(tài)與有狀態(tài)組件的分離無狀態(tài)組件路由決策器特別是模型推理部分、某些工具函數(shù)。這些組件可以輕松地水平擴(kuò)展用多個(gè)實(shí)例加負(fù)載均衡即可。有狀態(tài)組件循環(huán)引擎和上下文管理器是典型的有狀態(tài)組件。一個(gè)正在運(yùn)行的任務(wù)其循環(huán)狀態(tài)和上下文必須由同一個(gè)進(jìn)程或線程來維護(hù)不能隨意遷移。對此常見的模式是采用分片Sharding或基于會話的粘性Sticky Session。例如通過會話ID的哈希值決定由哪個(gè)后端實(shí)例來處理該會話的所有后續(xù)請求。水平擴(kuò)展循環(huán)引擎由于循環(huán)引擎有狀態(tài)擴(kuò)展起來更復(fù)雜。一種架構(gòu)是采用“主從”模式或“事件溯源”模式。事件溯源模式將Agent的每一次狀態(tài)變更如“路由決策A”、“執(zhí)行動作B成功”、“更新上下文C”都記錄為一個(gè)不可變的事件Event并持久化到事件流如Kafka中。循環(huán)引擎本身可以設(shè)計(jì)得相對輕量它讀取事件流根據(jù)當(dāng)前狀態(tài)和事件計(jì)算出新狀態(tài)并生成新的事件。這樣多個(gè)循環(huán)引擎實(shí)例可以消費(fèi)同一個(gè)事件流的不同分區(qū)狀態(tài)的計(jì)算邏輯是確定的最終狀態(tài)由事件序列決定便于擴(kuò)展和故障恢復(fù)。容錯(cuò)與高可用檢查點(diǎn)Checkpointing循環(huán)引擎定期將任務(wù)上下文和循環(huán)進(jìn)度持久化到可靠的存儲中。當(dāng)實(shí)例故障時(shí)調(diào)度器可以將該任務(wù)重新分配給另一個(gè)健康的實(shí)例新實(shí)例從最新的檢查點(diǎn)加載狀態(tài)并繼續(xù)執(zhí)行。優(yōu)雅降級當(dāng)路由模型服務(wù)或某個(gè)關(guān)鍵動作服務(wù)不可用時(shí)系統(tǒng)應(yīng)能降級到備用方案。例如路由降級到基于規(guī)則的簡單版本動作服務(wù)不可用則返回“服務(wù)暫不可用請稍后重試”并記錄任務(wù)狀態(tài)為“等待重試”。構(gòu)建一個(gè)健壯、高效的Agent運(yùn)行時(shí)系統(tǒng)是一個(gè)持續(xù)迭代和平衡的過程。從清晰理解循環(huán)、路由、上下文這三個(gè)核心機(jī)制開始在設(shè)計(jì)和編碼時(shí)始終想著狀態(tài)如何流轉(zhuǎn)、決策如何做出、信息如何保存就能避開很多深坑。