1. 項(xiàng)目概述從“筆記”到“工程實(shí)踐”最近在整理Harness Engineering相關(guān)的技術(shù)文檔時(shí)我發(fā)現(xiàn)關(guān)于Agent運(yùn)行時(shí)核心機(jī)制的討論尤其是循環(huán)、路由與上下文這三塊是很多開(kāi)發(fā)者從“會(huì)用框架”到“理解框架”的關(guān)鍵門(mén)檻。網(wǎng)上能找到的資料要么過(guò)于零散只講某個(gè)API怎么調(diào)用要么過(guò)于理論堆砌一堆架構(gòu)圖卻不說(shuō)清楚代碼到底怎么跑起來(lái)的。這讓我覺(jué)得有必要結(jié)合自己踩過(guò)的坑把這塊“硬骨頭”啃下來(lái)寫(xiě)一篇能直接指導(dǎo)編碼的深度解析。所謂Harness Engineering你可以把它理解為一套用于構(gòu)建、管理和控制智能體Agent的工程化框架與最佳實(shí)踐。它關(guān)注的不是某個(gè)單一的算法模型而是如何讓多個(gè)Agent協(xié)同、穩(wěn)定、高效地完成復(fù)雜任務(wù)。在這個(gè)體系里Agent運(yùn)行時(shí)就是那個(gè)讓智能體“活”起來(lái)、能夠感知、決策并執(zhí)行的核心引擎。而驅(qū)動(dòng)這個(gè)引擎的三大核心機(jī)制正是循環(huán)Loop、路由Routing和上下文Context。理解它們你才能真的駕馭Agent而不是被框架牽著鼻子走。這篇文章適合誰(shuí)呢如果你正在或打算開(kāi)發(fā)基于Agent的應(yīng)用無(wú)論是自動(dòng)化工作流、智能客服還是復(fù)雜的決策系統(tǒng)并且已經(jīng)過(guò)了“Hello World”階段開(kāi)始頭疼于Agent的狀態(tài)管理、任務(wù)調(diào)度和信息流轉(zhuǎn)問(wèn)題那么這里面的內(nèi)容就是為你準(zhǔn)備的。我會(huì)盡量避開(kāi)空泛的概念用具體的代碼段、設(shè)計(jì)抉擇背后的“為什么”以及我實(shí)際調(diào)試中總結(jié)出的“坑點(diǎn)”來(lái)展開(kāi)。2. 核心機(jī)制深度拆解循環(huán)、路由與上下文的角色與聯(lián)動(dòng)在深入細(xì)節(jié)之前我們必須建立一個(gè)頂層的認(rèn)知模型這三個(gè)機(jī)制不是孤立的它們共同構(gòu)成了Agent運(yùn)行時(shí)的一個(gè)完整“心跳周期”。想象一下一個(gè)處理用戶咨詢的客服Agent。它拿到用戶問(wèn)題上下文需要決定這個(gè)問(wèn)題屬于售前、售后還是技術(shù)故障路由然后根據(jù)決定調(diào)用相應(yīng)的子能力或工具去執(zhí)行執(zhí)行后產(chǎn)生新的結(jié)果并評(píng)估是否還需要進(jìn)一步追問(wèn)或可以結(jié)束循環(huán)。這個(gè)過(guò)程周而復(fù)始。2.1 循環(huán)LoopAgent執(zhí)行的狀態(tài)機(jī)與驅(qū)動(dòng)力循環(huán)機(jī)制是Agent運(yùn)行的“節(jié)拍器”。它決定了Agent在單次觸發(fā)后如何推進(jìn)其內(nèi)部狀態(tài)何時(shí)停止。最簡(jiǎn)單的循環(huán)就是“執(zhí)行一次就結(jié)束”O(jiān)ne-Shot。但對(duì)于復(fù)雜任務(wù)我們需要更強(qiáng)大的循環(huán)模式。2.1.1 常見(jiàn)循環(huán)模式與實(shí)踐選擇While-Loop條件循環(huán) 這是最常用、最靈活的循環(huán)模式。它基于一個(gè)或多個(gè)條件來(lái)決定是否繼續(xù)迭代。例如一個(gè)數(shù)據(jù)分析Agent可能會(huì)循環(huán)執(zhí)行“查詢-分析-提煉”步驟直到分析結(jié)果的置信度達(dá)到某個(gè)閾值或者用戶明確要求停止。# 偽代碼示例基于條件的While循環(huán) context initialize_context(user_query) while not should_stop(context): # 1. 路由決策根據(jù)當(dāng)前context決定下一步動(dòng)作 action routing_engine.decide(context) # 2. 執(zhí)行動(dòng)作 result execute_action(action, context) # 3. 更新上下文 context.update(result, action) # 4. 評(píng)估停止條件例如任務(wù)完成、達(dá)到最大步數(shù)、用戶中斷 context.evaluate_stop_condition()實(shí)操心得should_stop函數(shù)的設(shè)計(jì)是核心。不要只依賴單一條件如任務(wù)標(biāo)記完成最好結(jié)合多個(gè)維度最大迭代次數(shù)防止死循環(huán)、用戶主動(dòng)取消信號(hào)、連續(xù)多次迭代未產(chǎn)生有效進(jìn)展等。我通常會(huì)設(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ù)庫(kù)連接, 驗(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í)又能訪問(wèn)到必要的父級(jí)信息。需要清晰界定上下文的作用域和繼承關(guān)系否則很容易造成信息污染或丟失。2.1.2 循環(huán)控制的高級(jí)技巧暫停與恢復(fù)Pause/Resume對(duì)于長(zhǎng)時(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)來(lái)提升效率。但要注意共享上下文資源的線程安全問(wèn)題。一個(gè)穩(wěn)妥的做法是采用“復(fù)制-合并”策略每個(gè)并行任務(wù)處理上下文的一個(gè)副本執(zhí)行完畢后再將結(jié)果合并回主上下文。循環(huán)超時(shí)與看門(mén)狗Watchdog必須為每個(gè)循環(huán)設(shè)置超時(shí)機(jī)制。一個(gè)獨(dú)立的看門(mén)狗線程可以監(jiān)控主循環(huán)的執(zhí)行時(shí)間一旦超時(shí)就觸發(fā)中斷流程保存現(xiàn)場(chǎng)并上報(bào)錯(cuò)誤避免Agent“卡死”。2.2 路由RoutingAgent的決策中樞與流量控制器如果說(shuō)循環(huán)是“節(jié)奏”那么路由就是“方向”。它負(fù)責(zé)在運(yùn)行時(shí)根據(jù)當(dāng)前上下文動(dòng)態(tài)決定下一步該執(zhí)行哪個(gè)動(dòng)作、調(diào)用哪個(gè)工具、或者將任務(wù)移交給哪個(gè)更專業(yè)的子Agent。2.2.1 路由策略解析路由的核心是一個(gè)決策函數(shù)Input(Current_Context) - Output(Next_Action/Agent)?；谝?guī)則的路由Rule-Based 最簡(jiǎn)單直接的方式。通過(guò)預(yù)定義的if-else或決策樹(shù)來(lái)路由。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ù)靈活性差無(wú)法處理未預(yù)見(jiàn)的情況?；谀Ｐ偷穆酚蒑odel-Based 利用機(jī)器學(xué)習(xí)模型如分類器、甚至小型LLM來(lái)學(xué)習(xí)從上下文到最佳動(dòng)作的映射。這是當(dāng)前復(fù)雜Agent系統(tǒng)的趨勢(shì)。# 使用嵌入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ù)定義動(dòng)作及其描述 self.action_descriptions { search_db: 在知識(shí)庫(kù)中搜索相關(guān)信息, call_api: 調(diào)用外部API獲取實(shí)時(shí)數(shù)據(jù), generate_report: 生成分析總結(jié)報(bào)告 } # 預(yù)計(jì)算動(dòng)作描述的向量 self.action_vectors {name: self.model.encode(desc) for name, desc in self.action_descriptions.items()} def route(self, context): # 將當(dāng)前上下文如最新用戶問(wèn)題編碼為向量 query_vector self.model.encode(context.latest_query) # 計(jì)算與所有動(dòng)作的余弦相似度 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 # 返回最相似的動(dòng)作 return max(similarities, keysimilarities.get)實(shí)操心得模型路由的關(guān)鍵在于高質(zhì)量的動(dòng)作描述和上下文特征工程。動(dòng)作描述要精準(zhǔn)概括其功能和適用場(chǎng)景。上下文不能只扔進(jìn)模型需要提煉出關(guān)鍵特征如用戶意圖、對(duì)話歷史摘要、當(dāng)前任務(wù)階段等?；旌下酚蒆ybrid 結(jié)合規(guī)則和模型的優(yōu)勢(shì)。通常用規(guī)則處理明確、高優(yōu)先級(jí)的場(chǎng)景如安全攔截、緊急轉(zhuǎn)人工用模型處理復(fù)雜的、模糊的決策。也可以先用模型給出幾個(gè)候選再用規(guī)則進(jìn)行篩選和排序。2.2.2 路由表與策略鏈在工程實(shí)現(xiàn)上路由決策往往不是一步完成的。路由表Routing Table一個(gè)可配置的映射表將狀態(tài)或意圖映射到處理單元。適合規(guī)則路由便于運(yùn)營(yíng)人員修改。策略鏈Policy Chain按順序執(zhí)行多個(gè)路由策略。例如先經(jīng)過(guò)一個(gè)“過(guò)濾器策略”排除非法請(qǐng)求再經(jīng)過(guò)一個(gè)“優(yōu)先級(jí)策略”識(shí)別VIP用戶最后經(jīng)過(guò)一個(gè)“負(fù)載均衡策略”分配Agent實(shí)例。每個(gè)策略都可以對(duì)路由結(jié)果進(jìn)行修改或傳遞。注意路由決策本身也是有成本的。要避免在每次循環(huán)中都進(jìn)行非常復(fù)雜的模型推理。可以考慮對(duì)路由結(jié)果進(jìn)行緩存例如相同上下文指紋在短時(shí)間內(nèi)路由到相同目標(biāo)或者將路由決策的粒度放粗每N步或任務(wù)階段變更時(shí)才重新路由。2.3 上下文ContextAgent的“記憶”與“工作臺(tái)”上下文是貫穿循環(huán)和路由的“血液”。它承載了Agent任務(wù)執(zhí)行過(guò)程中的所有狀態(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)會(huì)話上下文Session Context范圍一次用戶會(huì)話可能包含多輪對(duì)話的全局信息。內(nèi)容用戶ID、會(huì)話ID、創(chuàng)建時(shí)間、全局配置、用戶長(zhǎng)期偏好、安全令牌等。生命周期從會(huì)話開(kāi)始到結(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í)行的動(dòng)作及其結(jié)果、臨時(shí)變量。生命周期一次循環(huán)開(kāi)始到結(jié)束。2.3.2 上下文的數(shù)據(jù)流與持久化上下文數(shù)據(jù)需要在循環(huán)、路由以及不同的處理單元間流動(dòng)。數(shù)據(jù)流設(shè)計(jì)建議采用不可變Immutable或?qū)憰r(shí)復(fù)制Copy-on-Write的理念。每次循環(huán)或動(dòng)作執(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ù)開(kāi)始、任務(wù)完成、每輪循環(huán)結(jié)束尤其是長(zhǎng)循環(huán)、以及發(fā)生錯(cuò)誤時(shí)。存儲(chǔ)什么至少需要存儲(chǔ)足以恢復(fù)任務(wù)狀態(tài)的最小數(shù)據(jù)集。包括上下文的核心字段、循環(huán)的進(jìn)度標(biāo)識(shí)、路由決策歷史等。存儲(chǔ)后端根據(jù)延遲和一致性要求選擇。Redis適合做高速緩存PostgreSQL/MongoDB適合做可靠存儲(chǔ)。復(fù)雜的上下文對(duì)象可能需要序列化如JSON、MessagePack、Pickle后存儲(chǔ)。實(shí)操心得為上下文對(duì)象設(shè)計(jì)一個(gè)版本號(hào)version字段。當(dāng)你的Agent代碼升級(jí)上下文結(jié)構(gòu)可能變化版本號(hào)可以幫助你進(jìn)行數(shù)據(jù)遷移或兼容性處理。2.3.3 上下文壓縮與摘要隨著對(duì)話或任務(wù)進(jìn)行上下文會(huì)不斷膨脹尤其是包含長(zhǎng)對(duì)話歷史或大量中間數(shù)據(jù)這會(huì)導(dǎo)致幾個(gè)問(wèn)題1超出模型token限制2降低路由和決策效率3增加存儲(chǔ)和傳輸開(kāi)銷。因此上下文壓縮Context Compression是高級(jí)Agent系統(tǒng)的必備技能。滑動(dòng)窗口只保留最近N輪對(duì)話。最簡(jiǎn)單但可能丟失關(guān)鍵早期信息。關(guān)鍵信息提取使用一個(gè)輕量級(jí)模型或規(guī)則從歷史上下文中提取出與當(dāng)前任務(wù)最相關(guān)的實(shí)體、意圖、事實(shí)結(jié)論丟棄冗余細(xì)節(jié)。增量式摘要在每輪或每N輪后動(dòng)態(tài)生成一個(gè)不斷更新的對(duì)話摘要。新的決策基于這個(gè)摘要和最新輸入而不是全部原始?xì)v史。# 偽代碼簡(jiǎn)單的增量摘要 class ConversationSummarizer: def __init__(self): self.summary def update_summary(self, new_dialogue_turn): # 將當(dāng)前摘要和新對(duì)話輪次一起讓LLM生成新的摘要 prompt f 現(xiàn)有摘要{self.summary} 最新一輪對(duì)話 用戶{new_dialogue_turn.user} Agent{new_dialogue_turn.agent} 請(qǐng)基于以上信息更新對(duì)話摘要保留所有關(guān)鍵決策、事實(shí)和待辦事項(xiàng)。 更新后的摘要 self.summary llm_invoke(prompt) return self.summary注意事項(xiàng)摘要的生成本身需要成本且可能存在信息損失。需要權(quán)衡摘要的更新頻率和精度。對(duì)于關(guān)鍵業(yè)務(wù)信息即使被摘要了也應(yīng)將其結(jié)構(gòu)化后單獨(dú)存儲(chǔ)在任務(wù)上下文中。3. 實(shí)戰(zhàn)構(gòu)建一個(gè)具備核心運(yùn)行機(jī)制的簡(jiǎn)易任務(wù)處理Agent理論說(shuō)再多不如動(dòng)手搭一個(gè)。我們來(lái)設(shè)計(jì)一個(gè)簡(jiǎn)易的“技術(shù)支持工單處理Agent”它需要理解用戶問(wèn)題自動(dòng)嘗試一些修復(fù)步驟如果不行則路由給人工客服。3.1 系統(tǒng)架構(gòu)與組件設(shè)計(jì)我們將系統(tǒng)分為以下模塊主循環(huán)引擎MainLoopEngine控制整體執(zhí)行流程。上下文管理器ContextManager創(chuàng)建、更新、持久化上下文。路由決策器Router基于上下文決定下一步。動(dòng)作執(zhí)行器ActionExecutor執(zhí)行具體的工具調(diào)用或子Agent任務(wù)。持久化存儲(chǔ)Storage用于保存上下文和會(huì)話狀態(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: 會(huì)話級(jí)上下文 session_id: str user_id: str created_at: datetime attributes: Dict[str, Any] field(default_factorydict) # 存放用戶偏好等 dataclass class TaskContext: 任務(wù)級(jí)上下文一個(gè)工單 task_id: str session_id: str description: str # 用戶問(wèn)題描述 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 # 最大自動(dòng)處理步數(shù) current_step: int 0 history: List[Dict] field(default_factorylist) # 記錄每一步操作 extracted_info: Dict[str, Any] field(default_factorydict) # 提取的關(guān)鍵信息如錯(cuò)誤代碼、設(shè)備型號(hào) dataclass class TurnContext: 回合級(jí)上下文 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: 自動(dòng)處理步數(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ù)輪次可能無(wú)新輸入 ) # 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í)行動(dòng)作 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動(dòng)作執(zhí)行失敗: {e} # 執(zhí)行失敗可以考慮重試或直接升級(jí) 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ú)存儲(chǔ) self.storage.save_turn(turn_ctx) def _update_task_from_result(self, task_ctx: TaskContext, result: Dict): 根據(jù)動(dòng)作執(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)先級(jí)規(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ī)則的快速路由明確場(chǎng)景 rule_based_action self.rule_router.decide(turn_ctx) if rule_based_action and rule_based_action ! unknown: return rule_based_action # 第三層基于模型的智能路由模糊場(chǎng)景 model_based_action self.model_router.decide(turn_ctx) if model_based_action: return model_based_action # 默認(rèn)降級(jí)方案 return self.fallback_action def _is_urgent_issue(self, turn_ctx): # 判斷是否為緊急問(wèn)題例如包含“宕機(jī)”、“無(wú)法使用”等關(guān)鍵詞且來(lái)自VIP用戶 urgent_keywords [宕機(jī), 崩潰, 完全無(wú)法] task_desc turn_ctx.task_context.description # 這里假設(shè)能從session_ctx獲取用戶等級(jí)簡(jiǎn)化處理 return any(kw in task_desc for kw in urgent_keywords) def _contains_sensitive_info(self, turn_ctx): # 簡(jiǎn)單示例檢測(cè)是否包含疑似密碼的信息 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 動(dòng)作執(zhí)行器的設(shè)計(jì)模式動(dòng)作執(zhí)行器負(fù)責(zé)將路由決策的“動(dòng)作名稱”轉(zhuǎn)化為具體的操作。一個(gè)好的模式是使用“注冊(cè)表”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未知?jiǎng)幼? {action_name}) try: # 執(zhí)行動(dòng)作并傳入當(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 # 模擬搜索知識(shí)庫(kù) 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跑起來(lái)只是第一步真正上線后各種意想不到的問(wèn)題才會(huì)浮現(xiàn)。下面分享幾個(gè)我實(shí)踐中遇到的典型挑戰(zhàn)和解決思路。4.1 循環(huán)失控與死鎖預(yù)防問(wèn)題場(chǎng)景一個(gè)用于處理文檔的Agent在“解析-提取-驗(yàn)證”的循環(huán)中因?yàn)槟硞€(gè)邊緣文檔格式解析異常導(dǎo)致驗(yàn)證永遠(yuǎn)無(wú)法通過(guò)循環(huán)卡死直到達(dá)到最大步數(shù)才超時(shí)退出浪費(fèi)資源且體驗(yàn)差。根因分析循環(huán)的停止條件過(guò)于依賴業(yè)務(wù)結(jié)果如“驗(yàn)證通過(guò)”未考慮異常狀態(tài)。缺乏對(duì)循環(huán)內(nèi)“無(wú)進(jìn)展”狀態(tài)的檢測(cè)。解決方案設(shè)置多維停止條件除了業(yè)務(wù)成功狀態(tài)和最大步數(shù)增加“無(wú)進(jìn)展檢測(cè)”。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: 檢測(cè)無(wú)進(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è)無(wú)效動(dòng)作或產(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引入看門(mén)狗Watchdog線程在主循環(huán)外啟動(dòng)一個(gè)監(jiān)控線程檢查單次循環(huán)的執(zhí)行時(shí)間。如果某次循環(huán)執(zhí)行時(shí)間遠(yuǎn)超歷史平均時(shí)間例如3個(gè)標(biāo)準(zhǔn)差以外則向主線程發(fā)送中斷信號(hào)保存當(dāng)前狀態(tài)并標(biāo)記為“超時(shí)異?！薄?.2 路由決策的準(zhǔn)確性與效率平衡問(wèn)題場(chǎng)景使用一個(gè)大型語(yǔ)言模型LLM作為路由決策器雖然準(zhǔn)確率高但每個(gè)請(qǐng)求都調(diào)用LLM導(dǎo)致響應(yīng)延遲高、成本昂貴。根因分析路由決策的粒度太細(xì)且未利用緩存。解決方案分層路由與緩存第一層意圖快速分類。使用一個(gè)輕量級(jí)文本分類模型如FastText、小規(guī)模BERT或關(guān)鍵詞匹配將請(qǐng)求分到幾個(gè)大的意圖桶如“查詢”、“操作”、“故障”。這一步速度極快可以覆蓋80%的常見(jiàn)請(qǐng)求。第二層桶內(nèi)精細(xì)路由。只有進(jìn)入“故障”等復(fù)雜桶的請(qǐng)求才觸發(fā)更精細(xì)的LLM路由或規(guī)則路由。同時(shí)對(duì)路由結(jié)果進(jìn)行緩存。緩存的Key可以是“意圖桶用戶輸入文本的哈?！被颉吧舷挛奶卣髦讣y”。設(shè)置合理的TTL如5分鐘在短時(shí)間內(nèi)相同問(wèn)題無(wú)需重復(fù)計(jì)算路由。路由決策預(yù)熱與異步更新對(duì)于已知的、高頻的任務(wù)模板可以在系統(tǒng)啟動(dòng)時(shí)或低峰期預(yù)計(jì)算其路由結(jié)果存入緩存。對(duì)于模型路由可以異步定期更新模型而不影響在線推理路徑。4.3 上下文膨脹與信息丟失的權(quán)衡問(wèn)題場(chǎng)景一個(gè)多輪對(duì)話Agent隨著對(duì)話輪次增加上下文越來(lái)越長(zhǎng)最終超出模型Token限制。使用簡(jiǎn)單的滑動(dòng)窗口又丟失了對(duì)話開(kāi)頭約定的關(guān)鍵約束條件。根因分析上下文管理策略過(guò)于簡(jiǎn)單對(duì)所有信息一視同仁。解決方案結(jié)構(gòu)化上下文與重要性標(biāo)注不要將所有歷史都作為非結(jié)構(gòu)化的文本堆砌。設(shè)計(jì)結(jié)構(gòu)化的上下文對(duì)象將信息分類系統(tǒng)指令高優(yōu)先級(jí)Agent的角色、核心約束、對(duì)話目標(biāo)。這部分應(yīng)始終保留。關(guān)鍵事實(shí)與參數(shù)中優(yōu)先級(jí)用戶明確提供的實(shí)體、數(shù)字、選擇等?？梢蕴崛〕鰜?lái)放在一個(gè)“事實(shí)表”中。對(duì)話歷史低優(yōu)先級(jí)具體的每一輪問(wèn)答。這部分是壓縮的主要對(duì)象。動(dòng)態(tài)摘要與關(guān)鍵信息提取在每輪對(duì)話后不是保存全部原文而是用一個(gè)小模型或提示詞工程生成一個(gè)增量摘要。這個(gè)摘要會(huì)融合上一輪的摘要和本輪的新內(nèi)容。同時(shí)運(yùn)行一個(gè)命名實(shí)體識(shí)別NER或信息提取流程將本輪出現(xiàn)的新關(guān)鍵信息如訂單號(hào)、日期、產(chǎn)品型號(hào)提取出來(lái)添加到“事實(shí)表”中。后續(xù)的路由和動(dòng)作執(zhí)行主要參考“系統(tǒng)指令”、“事實(shí)表”和“最新摘要”而非全部歷史。只有當(dāng)需要深度理解某段歷史時(shí)才去查詢?cè)敿?xì)的、經(jīng)過(guò)索引的原始記錄如果保存了的話。class CompressingContextManager: def __init__(self, max_raw_turns10): self.system_instructions self.fact_table {} # 鍵值對(duì)存儲(chǔ)關(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. 保存原始記錄滑動(dòng)窗口 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新對(duì)話用戶說(shuō){user_input}Agent回復(fù){agent_response}\n請(qǐng)生成更新的摘要。 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)試與可觀測(cè)性建設(shè)當(dāng)Agent行為不符合預(yù)期時(shí)如何快速定位是循環(huán)、路由還是上下文的問(wèn)題結(jié)構(gòu)化日志不要在代碼里隨意打印print。為每個(gè)循環(huán)步驟、路由決策、上下文更新、動(dòng)作執(zhí)行記錄結(jié)構(gòu)化的日志。日志應(yīng)包含時(shí)間戳、會(huì)話ID、任務(wù)ID、步驟ID、組件名如Router、ActionX、日志級(jí)別、關(guān)鍵數(shù)據(jù)如路由決策結(jié)果、上下文快照的哈希、動(dòng)作執(zhí)行耗時(shí)和錯(cuò)誤信息。# 使用結(jié)構(gòu)化日志庫(kù)如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è)用戶請(qǐng)求注入追蹤ID。這個(gè)ID貫穿整個(gè)調(diào)用鏈從接收請(qǐng)求到循環(huán)的每一步到路由到子服務(wù)調(diào)用讓你能在儀表盤(pán)上清晰地看到一個(gè)請(qǐng)求的完整生命周期和耗時(shí)瓶頸。上下文快照與回放將每個(gè)關(guān)鍵步驟尤其是路由決策前后和動(dòng)作執(zhí)行前后的完整上下文對(duì)象序列化后存儲(chǔ)到可查詢的存儲(chǔ)中如Elasticsearch。當(dāng)出現(xiàn)問(wèn)題時(shí)你可以通過(guò)任務(wù)ID檢索出完整的上下文歷史在本地或測(cè)試環(huán)境“回放”該任務(wù)精確復(fù)現(xiàn)問(wèn)題。路由決策的可解釋性對(duì)于基于模型的路由不僅要輸出決策還要輸出置信度分?jǐn)?shù)和如果可能主要依據(jù)的特征。這能幫助你在調(diào)試時(shí)判斷是模型不準(zhǔn)還是輸入特征有問(wèn)題。5. 性能優(yōu)化與擴(kuò)展性考量當(dāng)你的Agent系統(tǒng)從原型走向生產(chǎn)服務(wù)大量并發(fā)用戶時(shí)性能和擴(kuò)展性就成為必須面對(duì)的問(wèn)題。5.1 運(yùn)行時(shí)性能優(yōu)化循環(huán)異步化如果Agent的某些動(dòng)作是I/O密集型的如調(diào)用外部API、查詢數(shù)據(jù)庫(kù)不要讓主循環(huán)同步等待?？梢詫?dòng)作提交到任務(wù)隊(duì)列如Celery、RabbitMQ主循環(huán)只負(fù)責(zé)生成任務(wù)和監(jiān)聽(tīng)結(jié)果。這樣單個(gè)Agent實(shí)例可以同時(shí)管理多個(gè)任務(wù)的執(zhí)行流極大提高吞吐量。路由決策批處理對(duì)于請(qǐng)求量大的場(chǎng)景可以將短時(shí)間內(nèi)的一批路由請(qǐng)求收集起來(lái)批量發(fā)送給模型進(jìn)行推理如果模型支持批量推理這比逐個(gè)請(qǐng)求效率高得多。上下文緩存會(huì)話級(jí)和任務(wù)級(jí)的上下文是高頻訪問(wèn)對(duì)象。使用內(nèi)存緩存如Redis存儲(chǔ)活躍的上下文對(duì)象避免每次循環(huán)都從數(shù)據(jù)庫(kù)讀取。注意設(shè)計(jì)合理的緩存失效和回寫(xiě)策略。動(dòng)作執(zhí)行結(jié)果緩存對(duì)于一些冪等的、結(jié)果相對(duì)穩(wěn)定的動(dòng)作如“根據(jù)城市名查詢天氣”可以對(duì)其結(jié)果進(jìn)行緩存。路由決策后先查緩存命中則直接返回避免重復(fù)執(zhí)行。5.2 系統(tǒng)擴(kuò)展性設(shè)計(jì)無(wú)狀態(tài)與有狀態(tài)組件的分離無(wú)狀態(tài)組件路由決策器特別是模型推理部分、某些工具函數(shù)。這些組件可以輕松地水平擴(kuò)展用多個(gè)實(shí)例加負(fù)載均衡即可。有狀態(tài)組件循環(huán)引擎和上下文管理器是典型的有狀態(tài)組件。一個(gè)正在運(yùn)行的任務(wù)其循環(huán)狀態(tài)和上下文必須由同一個(gè)進(jìn)程或線程來(lái)維護(hù)不能隨意遷移。對(duì)此常見(jiàn)的模式是采用分片Sharding或基于會(huì)話的粘性Sticky Session。例如通過(guò)會(huì)話ID的哈希值決定由哪個(gè)后端實(shí)例來(lái)處理該會(huì)話的所有后續(xù)請(qǐng)求。水平擴(kuò)展循環(huán)引擎由于循環(huán)引擎有狀態(tài)擴(kuò)展起來(lái)更復(fù)雜。一種架構(gòu)是采用“主從”模式或“事件溯源”模式。事件溯源模式將Agent的每一次狀態(tài)變更如“路由決策A”、“執(zhí)行動(dòng)作B成功”、“更新上下文C”都記錄為一個(gè)不可變的事件Event并持久化到事件流如Kafka中。循環(huán)引擎本身可以設(shè)計(jì)得相對(duì)輕量它讀取事件流根據(jù)當(dāng)前狀態(tài)和事件計(jì)算出新?tīng)顟B(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)度持久化到可靠的存儲(chǔ)中。當(dāng)實(shí)例故障時(shí)調(diào)度器可以將該任務(wù)重新分配給另一個(gè)健康的實(shí)例新實(shí)例從最新的檢查點(diǎn)加載狀態(tài)并繼續(xù)執(zhí)行。優(yōu)雅降級(jí)當(dāng)路由模型服務(wù)或某個(gè)關(guān)鍵動(dòng)作服務(wù)不可用時(shí)系統(tǒng)應(yīng)能降級(jí)到備用方案。例如路由降級(jí)到基于規(guī)則的簡(jiǎn)單版本動(dòng)作服務(wù)不可用則返回“服務(wù)暫不可用請(qǐng)稍后重試”并記錄任務(wù)狀態(tài)為“等待重試”。構(gòu)建一個(gè)健壯、高效的Agent運(yùn)行時(shí)系統(tǒng)是一個(gè)持續(xù)迭代和平衡的過(guò)程。從清晰理解循環(huán)、路由、上下文這三個(gè)核心機(jī)制開(kāi)始在設(shè)計(jì)和編碼時(shí)始終想著狀態(tài)如何流轉(zhuǎn)、決策如何做出、信息如何保存就能避開(kāi)很多深坑。