1. 從后端工程師的視角看Agent不只是“智能體”最近“Agent”這個(gè)詞火得不行各種AI智能體、自動(dòng)化代理的框架和教程層出不窮。作為一個(gè)在后端領(lǐng)域摸爬滾打了十來(lái)年的老碼農(nóng)我最初看到這些概念時(shí)第一反應(yīng)其實(shí)是有點(diǎn)“懵”的。什么“智能體”、“自主決策”、“任務(wù)分解”聽(tīng)起來(lái)很高級(jí)但落到代碼層面它到底是個(gè)啥難道不是我們后端天天在寫的那些服務(wù)、調(diào)度器和狀態(tài)機(jī)嗎后來(lái)我花了不少時(shí)間去研究從LangChain到AutoGPT再到一些開(kāi)源的Agent框架。我發(fā)現(xiàn)用后端開(kāi)發(fā)的思維去理解Agent事情會(huì)變得異常清晰和親切。Agent的核心本質(zhì)上就是一個(gè)具備特定能力、能感知環(huán)境、能執(zhí)行動(dòng)作并處理反饋的“服務(wù)進(jìn)程”。這不就是我們后端系統(tǒng)里一個(gè)設(shè)計(jì)良好的微服務(wù)或者一個(gè)后臺(tái)任務(wù)調(diào)度器該有的樣子嗎今天我們不談那些高大上的理論就從最樸素的“后端思維”出發(fā)把Agent這個(gè)黑盒子拆開(kāi)看看里面的核心架構(gòu)到底是怎么運(yùn)轉(zhuǎn)的。我會(huì)重點(diǎn)聚焦在三個(gè)我認(rèn)為最核心、也最體現(xiàn)后端工程化思想的組件上三元組狀態(tài)管理、工具調(diào)用服務(wù)編排和錯(cuò)誤處理系統(tǒng)韌性。你會(huì)發(fā)現(xiàn)理解了這三塊你不僅能看懂Agent甚至能自己動(dòng)手搭一個(gè)更健壯、更可控的Agent出來(lái)。2. 核心基石用“三元組”思維重構(gòu)Agent的狀態(tài)管理在分布式系統(tǒng)和復(fù)雜業(yè)務(wù)邏輯中我們?nèi)绾蚊枋鲆粋€(gè)實(shí)體的當(dāng)前狀況后端常用的方法是定義狀態(tài)機(jī)State Machine或者使用一個(gè)包含關(guān)鍵屬性的上下文對(duì)象Context Object。Agent的“三元組”思想與此異曲同工它提供了一個(gè)極其簡(jiǎn)潔而強(qiáng)大的模型來(lái)刻畫Agent在任一時(shí)刻的“快照”。2.1 什么是Agent的三元組簡(jiǎn)單來(lái)說(shuō)Agent的三元組指的是(Thought, Action, Observation)。這不是一個(gè)冰冷的理論而是一個(gè)動(dòng)態(tài)的工作循環(huán)單元。Thought思考這是Agent的“內(nèi)部狀態(tài)”或“決策依據(jù)”。它不是一個(gè)模糊的想法而應(yīng)該是一個(gè)結(jié)構(gòu)化的數(shù)據(jù)?？梢岳斫鉃楹蠖朔?wù)處理請(qǐng)求時(shí)的“內(nèi)部上下文”或“決策邏輯的輸出”。例如一個(gè)訂票Agent的Thought可能是{“user_intent”: “book_flight”, “parsed_info”: {“departure”: “上?！? “arrival”: “北京”, “date”: “2023-10-01”}, “next_step”: “search_flights”}。在后端這類似于一個(gè)工作流引擎中當(dāng)前任務(wù)節(jié)點(diǎn)的輸出和傳遞給下一個(gè)節(jié)點(diǎn)的參數(shù)。Action行動(dòng)基于ThoughtAgent決定要執(zhí)行的具體操作。這個(gè)操作通常是調(diào)用一個(gè)“工具”Tool。例如對(duì)應(yīng)的Action可能是{“tool_name”: “flight_search_api”, “parameters”: {“from”: “上?！? “to”: “北京”, “date”: “2023-10-01”}}。這完全對(duì)應(yīng)了后端服務(wù)中的一個(gè)“服務(wù)調(diào)用”或“API調(diào)用”指令。Observation觀察執(zhí)行Action后從環(huán)境或工具返回的結(jié)果。例如調(diào)用航班搜索API后返回的JSON數(shù)據(jù){“flights”: […], “status”: “success”}。在后端這就是一個(gè)下游服務(wù)或數(shù)據(jù)庫(kù)查詢的響應(yīng)。為什么這個(gè)三元組如此重要因?yàn)樗鼘gent不可控的“思考”過(guò)程變成了一個(gè)可觀測(cè)、可記錄、可調(diào)試的數(shù)據(jù)流。每一次循環(huán)Thought - Action - Observation都會(huì)產(chǎn)生一條日志這讓我們可以像排查后端接口調(diào)用鏈一樣去追溯Agent的決策路徑。2.2 后端工程中的三元組實(shí)踐在實(shí)際的后端架構(gòu)中我們其實(shí)一直在不自覺(jué)地使用類似三元組的概念。在消息隊(duì)列消費(fèi)者中消費(fèi)者從隊(duì)列取到一條消息Observation經(jīng)過(guò)業(yè)務(wù)邏輯處理Thought然后可能調(diào)用另一個(gè)服務(wù)或?qū)懭霐?shù)據(jù)庫(kù)Action并等待結(jié)果新的Observation來(lái)決定是確認(rèn)消息還是重試。在工作流引擎中每一個(gè)節(jié)點(diǎn)接收輸入Observation執(zhí)行節(jié)點(diǎn)邏輯Thought調(diào)用某個(gè)服務(wù)或腳本Action并將輸出作為下一個(gè)節(jié)點(diǎn)的輸入新的Observation。在狀態(tài)機(jī)中當(dāng)前狀態(tài)可以類比為Thought的某種聚合和接收的事件Observation共同決定要執(zhí)行的動(dòng)作Action和遷移到的下一個(gè)狀態(tài)新的Thought。實(shí)操心得如何設(shè)計(jì)一個(gè)好的Thought這是三元組里最需要設(shè)計(jì)的地方。Thought不能太“重”包含所有歷史信息臃腫低效也不能太“輕”信息不足無(wú)法決策。我的經(jīng)驗(yàn)是結(jié)構(gòu)化堅(jiān)決使用JSON等結(jié)構(gòu)化數(shù)據(jù)而不是自然語(yǔ)言描述。這便于程序解析和傳遞。聚焦當(dāng)前目標(biāo)Thought應(yīng)清晰包含“我當(dāng)前要解決什么問(wèn)題”和“我已經(jīng)掌握了什么信息”。包含元信息可以加入step_count當(dāng)前是第幾步、max_steps最大步數(shù)限制、retry_count對(duì)當(dāng)前Action的重試次數(shù)等控制信息防止Agent陷入死循環(huán)。提示在實(shí)現(xiàn)時(shí)可以將整個(gè)三元組序列一個(gè)對(duì)話或任務(wù)的歷史作為“上下文”Context保存下來(lái)這類似于后端會(huì)話Session或業(yè)務(wù)流程實(shí)例Process Instance的概念。3. 核心能力將“工具調(diào)用”視為服務(wù)編排與API聚合Agent自己并不創(chuàng)造新能力它的強(qiáng)大在于能夠“使用工具”。這和后端開(kāi)發(fā)中的“服務(wù)編排”Service Orchestration和“API聚合”API Gateway思想完全一致。一個(gè)復(fù)雜的業(yè)務(wù)接口后端可能會(huì)調(diào)用用戶服務(wù)、訂單服務(wù)、支付服務(wù)等多個(gè)下游接口來(lái)完成。Agent也是如此它通過(guò)組合調(diào)用各種工具可以視為一個(gè)個(gè)微服務(wù)來(lái)完成復(fù)雜任務(wù)。3.1 工具Tool的本質(zhì)標(biāo)準(zhǔn)化的能力接口一個(gè)設(shè)計(jì)良好的工具應(yīng)該像一個(gè)設(shè)計(jì)良好的API接口明確的輸入輸出必須有清晰的參數(shù)定義和返回格式。例如一個(gè)calculator工具輸入是{“expression”: “string”}輸出是{“result”: “number”}。功能單一一個(gè)工具只做一件事并且做好。search_web工具就負(fù)責(zé)搜索send_email工具就負(fù)責(zé)發(fā)郵件。這符合微服務(wù)的“單一職責(zé)”原則?？砂l(fā)現(xiàn)性Agent需要知道有哪些工具可用。這通常通過(guò)一個(gè)“工具注冊(cè)表”或“工具描述清單”來(lái)實(shí)現(xiàn)。清單里會(huì)包含工具的名稱、描述、參數(shù)Schema等。這就像后端的服務(wù)注冊(cè)中心如Eureka, Nacos。// 一個(gè)工具的描述示例 { “name”: “get_weather”, “description”: “獲取指定城市的當(dāng)前天氣信息”, “parameters”: { “type”: “object”, “properties”: { “city”: { “type”: “string”, “description”: “城市名稱例如北京” } }, “required”: [“city”] } }3.2 工具調(diào)用的流程從意圖到執(zhí)行當(dāng)Agent決定要調(diào)用一個(gè)工具時(shí)其內(nèi)部流程和后端網(wǎng)關(guān)調(diào)用下游服務(wù)非常相似意圖解析與工具選擇Agent根據(jù)Thought決定需要什么能力如“用戶想知道天氣”然后從工具注冊(cè)表中匹配最合適的工具get_weather。這類似于后端網(wǎng)關(guān)根據(jù)路由規(guī)則選擇目標(biāo)服務(wù)。參數(shù)組裝根據(jù)Thought中的信息和工具的參數(shù)定義組裝出完整的調(diào)用參數(shù){“city”: “北京”}。這里可能涉及參數(shù)轉(zhuǎn)換和默認(rèn)值填充。安全與權(quán)限校驗(yàn)在執(zhí)行調(diào)用前應(yīng)檢查當(dāng)前Agent是否有權(quán)限調(diào)用此工具參數(shù)是否安全如防止SQL注入或惡意請(qǐng)求。這是后端API網(wǎng)關(guān)的核心功能之一。執(zhí)行調(diào)用以同步或異步方式實(shí)際調(diào)用工具。工具可能是一個(gè)本地函數(shù)、一個(gè)HTTP API、一個(gè)數(shù)據(jù)庫(kù)查詢甚至是對(duì)另一個(gè)大模型的請(qǐng)求。結(jié)果標(biāo)準(zhǔn)化接收工具的原始返回并將其處理成Agent能理解的標(biāo)準(zhǔn)化Observation格式。如果工具調(diào)用失敗超時(shí)、錯(cuò)誤等也需要生成一個(gè)代表“錯(cuò)誤”的Observation。踩坑實(shí)錄工具調(diào)用的常見(jiàn)問(wèn)題工具描述不準(zhǔn)工具的描述或參數(shù)Schema寫得不清楚導(dǎo)致Agent無(wú)法正確調(diào)用。這就像API文檔寫錯(cuò)了調(diào)用方必然出錯(cuò)。解決方案像寫API文檔一樣嚴(yán)格定義工具并使用JSON Schema進(jìn)行校驗(yàn)。工具不可用或超時(shí)這是分布式系統(tǒng)的常態(tài)。解決方案必須為工具調(diào)用設(shè)置合理的超時(shí)時(shí)間和重試機(jī)制。在Observation中明確區(qū)分“工具返回業(yè)務(wù)錯(cuò)誤”和“工具調(diào)用本身失敗”。工具組合的復(fù)雜性當(dāng)需要連續(xù)調(diào)用多個(gè)有依賴關(guān)系的工具時(shí)邏輯會(huì)變得復(fù)雜。解決方案借鑒后端工作流引擎的思路在Thought中顯式地管理子任務(wù)狀態(tài)和依賴關(guān)系而不是依賴大模型一次性規(guī)劃所有步驟。4. 生命線像構(gòu)建分布式系統(tǒng)一樣設(shè)計(jì)Agent的錯(cuò)誤處理任何不處理錯(cuò)誤的系統(tǒng)都是玩具。對(duì)于Agent這種需要自主運(yùn)行、可能調(diào)用多個(gè)不穩(wěn)定外部服務(wù)的系統(tǒng)來(lái)說(shuō)健壯的錯(cuò)誤處理不是加分項(xiàng)而是生命線。后端開(kāi)發(fā)中我們處理服務(wù)超時(shí)、重試、降級(jí)、熔斷的經(jīng)驗(yàn)在這里可以完全復(fù)用。4.1 Agent可能遇到的四層錯(cuò)誤思考層錯(cuò)誤大模型本身“胡言亂語(yǔ)”輸出了無(wú)法解析的Thought或者做出了明顯不合理、不合規(guī)的決策。例如Thought中指定的tool_name在注冊(cè)表中不存在。工具調(diào)用層錯(cuò)誤這是最普遍的錯(cuò)誤包括網(wǎng)絡(luò)超時(shí)、接口返回4xx/5xx錯(cuò)誤、返回?cái)?shù)據(jù)格式不符合預(yù)期等。觀察解析層錯(cuò)誤工具返回的結(jié)果Observation可能非常復(fù)雜或混亂Agent在解析和理解這些結(jié)果時(shí)可能出現(xiàn)偏差。流程控制層錯(cuò)誤Agent陷入死循環(huán)不斷重復(fù)相同操作、任務(wù)長(zhǎng)時(shí)間無(wú)進(jìn)展、或違反了預(yù)設(shè)的業(yè)務(wù)規(guī)則如預(yù)算超標(biāo)。4.2 構(gòu)建錯(cuò)誤處理防線監(jiān)控、重試與熔斷第一道防線輸入輸出校驗(yàn)與結(jié)構(gòu)化在Thought生成后、Action執(zhí)行前加入一層校驗(yàn)。確保Action中的工具名、參數(shù)格式是合法的。這類似于API網(wǎng)關(guān)的參數(shù)驗(yàn)證?？梢允褂幂p量級(jí)的規(guī)則引擎或簡(jiǎn)單的模式匹配來(lái)實(shí)現(xiàn)。第二道防線工具調(diào)用的彈性策略重試對(duì)于網(wǎng)絡(luò)抖動(dòng)或瞬時(shí)故障導(dǎo)致的失敗應(yīng)立即進(jìn)行重試。重試策略如指數(shù)退避至關(guān)重要避免雪崩。超時(shí)控制每個(gè)工具調(diào)用必須有獨(dú)立的超時(shí)設(shè)置防止一個(gè)慢工具拖死整個(gè)Agent。熔斷與降級(jí)如果某個(gè)工具連續(xù)失敗可以暫時(shí)將其“熔斷”標(biāo)記為不可用并在Thought中反映這一點(diǎn)。Agent可以選擇備用工具或向用戶返回“該服務(wù)暫不可用”的友好提示。第三道防線流程看門狗這是Agent獨(dú)有的、非常重要的機(jī)制。你需要一個(gè)獨(dú)立的“看門狗”邏輯來(lái)監(jiān)控Agent的運(yùn)行狀態(tài)最大步數(shù)限制防止無(wú)限循環(huán)。在Thought中維護(hù)step_count超過(guò)閾值則強(qiáng)制終止任務(wù)并生成一個(gè)“任務(wù)超時(shí)”的最終Observation。重復(fù)動(dòng)作檢測(cè)記錄最近N次Action如果發(fā)現(xiàn)完全相同的Action被重復(fù)執(zhí)行且未推動(dòng)任務(wù)進(jìn)展應(yīng)中斷并嘗試替代方案。關(guān)鍵狀態(tài)檢查在每一步之后檢查是否違反了核心業(yè)務(wù)約束如“總花費(fèi)不能超過(guò)100元”。第四道防線優(yōu)雅的失敗與用戶交互不是所有錯(cuò)誤都需要Agent在后臺(tái)死磕。對(duì)于某些錯(cuò)誤尤其是經(jīng)過(guò)重試和降級(jí)后仍無(wú)法解決的最好的策略是“向上匯報(bào)”——即生成一個(gè)請(qǐng)求人類協(xié)助的Observation。例如“調(diào)用支付網(wǎng)關(guān)失敗原因可能是網(wǎng)絡(luò)問(wèn)題。已重試3次。是否嘗試其他支付方式或取消訂單” 這需要你在設(shè)計(jì)Agent的交互流程時(shí)就預(yù)留“人工接管”的接口。一個(gè)錯(cuò)誤處理的代碼片段思路class RobustAgent: def execute_action(self, action): tool_name action[“tool_name”] params action[“parameters”] # 1. 校驗(yàn) if tool_name not in self.tool_registry: return self._create_error_observation(f“未知工具{tool_name}”) # 2. 帶重試和超時(shí)的調(diào)用 max_retries 3 for attempt in range(max_retries): try: # 設(shè)置單次調(diào)用超時(shí) result self._call_tool_with_timeout(tool_name, params, timeout10) # 3. 結(jié)果標(biāo)準(zhǔn)化 return self._standardize_observation(result) except TimeoutError: if attempt max_retries - 1: return self._create_error_observation(f“工具{tool_name}調(diào)用超時(shí)”) time.sleep(2 ** attempt) # 指數(shù)退避 except ToolExecutionError as e: # 工具返回的業(yè)務(wù)錯(cuò)誤直接封裝 return self._create_error_observation(f“工具執(zhí)行失敗{str(e)}”) except Exception as e: # 其他未預(yù)料錯(cuò)誤 log.error(f“調(diào)用工具{tool_name}異?！? exc_infoe) if attempt max_retries - 1: return self._create_error_observation(“系統(tǒng)內(nèi)部錯(cuò)誤請(qǐng)稍后重試”)5. 實(shí)戰(zhàn)搭建一個(gè)具備后端工程化思想的簡(jiǎn)易Agent框架理論說(shuō)再多不如動(dòng)手搭一個(gè)。我們不依賴復(fù)雜的LangChain就用最樸素的Python代碼實(shí)現(xiàn)一個(gè)具備上述核心思想的Agent骨架。你會(huì)發(fā)現(xiàn)它和一個(gè)簡(jiǎn)單的異步服務(wù)編排引擎非常像。5.1 定義核心數(shù)據(jù)模型首先我們定義清晰的數(shù)據(jù)結(jié)構(gòu)這是所有穩(wěn)定后端系統(tǒng)的基礎(chǔ)。from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class Thought(BaseModel): 思考狀態(tài)結(jié)構(gòu)化的內(nèi)部上下文 goal: str Field(description“當(dāng)前要完成的總目標(biāo)”) step: int Field(default1, description“當(dāng)前步驟序號(hào)”) max_steps: int Field(default10, description“最大允許步驟數(shù)”) known_info: Dict[str, Any] Field(default_factorydict, description“已收集到的信息”) next_action_intent: str Field(description“下一步打算做什么的意圖描述”) class Action(BaseModel): 要執(zhí)行的動(dòng)作即調(diào)用哪個(gè)工具、傳什么參數(shù) tool_name: str parameters: Dict[str, Any] class Observation(BaseModel): 執(zhí)行動(dòng)作后的觀察結(jié)果 status: str # “success”, “error”, “need_human” data: Optional[Dict[str, Any]] None message: Optional[str] None # 成功或錯(cuò)誤的詳細(xì)信息 class AgentStep(BaseModel): 一個(gè)完整的三元組步驟記錄用于日志和調(diào)試 step_id: int thought: Thought action: Optional[Action] None observation: Optional[Observation] None5.2 實(shí)現(xiàn)工具管理器工具管理器扮演著服務(wù)注冊(cè)中心的角色。import inspect from functools import wraps class ToolRegistry: def __init__(self): self._tools: Dict[str, dict] {} def register(self, func): 裝飾器將函數(shù)注冊(cè)為工具 sig inspect.signature(func) params_schema { “type”: “object”, “properties”: {}, “required”: [] } for name, param in sig.parameters.items(): param_info {“type”: “string”} # 簡(jiǎn)化處理實(shí)際應(yīng)根據(jù)annotation生成 params_schema[“properties”][name] param_info if param.default inspect.Parameter.empty: params_schema[“required”].append(name) self._tools[func.__name__] { “function”: func, “description”: func.__doc__ or “”, “parameters_schema”: params_schema } wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper async def execute(self, tool_name: str, parameters: Dict) - Observation: if tool_name not in self._tools: return Observation(status“error”, messagef“Tool ‘{tool_name}’ not found.”) tool_info self._tools[tool_name] func tool_info[“function”] try: # 這里可以加入?yún)?shù)校驗(yàn)根據(jù)schema、權(quán)限檢查等 result await func(**parameters) if inspect.iscoroutinefunction(func) else func(**parameters) return Observation(status“success”, data{“result”: result}) except Exception as e: # 這里可以區(qū)分不同類型的異常進(jìn)行更精細(xì)的錯(cuò)誤處理 return Observation(status“error”, messagef“Tool execution failed: {str(e)}”) # 示例工具定義 tool_registry ToolRegistry() tool_registry.register def get_current_time(timezone: str “UTC”) - str: “”“獲取指定時(shí)區(qū)的當(dāng)前時(shí)間。”“” from datetime import datetime import pytz tz pytz.timezone(timezone) return datetime.now(tz).strftime(“%Y-%m-%d %H:%M:%S”) tool_registry.register def calculator(expression: str) - float: “”“計(jì)算一個(gè)簡(jiǎn)單的數(shù)學(xué)表達(dá)式。支持 , -, *, /?！薄啊?# 警告實(shí)際生產(chǎn)環(huán)境必須對(duì)表達(dá)式做嚴(yán)格的安全過(guò)濾防止代碼注入 allowed_chars set(“0123456789-*/(). ”) if not all(c in allowed_chars for c in expression): raise ValueError(“Expression contains invalid characters”) try: return eval(expression) except Exception as e: raise ValueError(f“Calculation error: {str(e)}”)5.3 實(shí)現(xiàn)Agent核心循環(huán)與錯(cuò)誤處理這是Agent的“大腦”和“神經(jīng)系統(tǒng)”包含了狀態(tài)管理和錯(cuò)誤處理的核心邏輯。class SimpleAgent: def __init__(self, tool_registry: ToolRegistry, llm_client): # llm_client是一個(gè)模擬的大語(yǔ)言模型調(diào)用接口 self.tool_registry tool_registry self.llm llm_client self.history: List[AgentStep] [] async def run(self, initial_goal: str) - Dict[str, Any]: 運(yùn)行Agent直到完成任務(wù)或達(dá)到最大步數(shù)。 current_thought Thought( goalinitial_goal, step1, known_info{}, next_action_intent“分析任務(wù)目標(biāo)” ) while current_thought.step current_thought.max_steps: print(f“\n Step {current_thought.step} ) print(f“Thought: {current_thought.next_action_intent}”) # 1. 基于當(dāng)前Thought決定下一步Action (這里用LLM或規(guī)則引擎) action await self._decide_action(current_thought) if action is None: # 決策失敗可能是LLM返回了無(wú)法解析的內(nèi)容 return {“status”: “error”, “message”: “Failed to decide next action.”} print(f“Action: Call {action.tool_name} with {action.parameters}”) # 2. 執(zhí)行Action獲取Observation observation await self.tool_registry.execute(action.tool_name, action.parameters) print(f“Observation: {observation.status} - {observation.message or observation.data}”) # 3. 記錄這一步 step_record AgentStep( step_idcurrent_thought.step, thoughtcurrent_thought.model_copy(), actionaction, observationobservation ) self.history.append(step_record) # 4. 檢查是否成功、是否需要人工介入、是否出錯(cuò) if observation.status “need_human”: return {“status”: “pending_human”, “history”: self.history, “message”: observation.message} elif observation.status “error”: # 錯(cuò)誤處理根據(jù)策略決定重試、換方案還是失敗 if not await self._handle_error(current_thought, action, observation): return {“status”: “failed”, “history”: self.history, “error”: observation.message} # 如果handle_error返回True表示已處理如重試?yán)^續(xù)循環(huán) continue # 5. 基于Observation和當(dāng)前狀態(tài)生成下一步的Thought (這里用LLM或規(guī)則引擎) next_thought await self._update_thought(current_thought, action, observation) if next_thought is None: return {“status”: “error”, “message”: “Failed to update thought.”} # 6. 檢查任務(wù)是否完成例如Thought中標(biāo)記goal_achieved為True if getattr(next_thought, “goal_achieved”, False): return {“status”: “success”, “history”: self.history, “result”: next_thought.known_info} current_thought next_thought current_thought.step 1 # 循環(huán)結(jié)束達(dá)到最大步數(shù) return {“status”: “max_steps_exceeded”, “history”: self.history} async def _decide_action(self, thought: Thought) - Optional[Action]: 決策邏輯。這里簡(jiǎn)化處理實(shí)際應(yīng)集成LLM或復(fù)雜的規(guī)則引擎。 # 示例一個(gè)非常簡(jiǎn)單的基于意圖的規(guī)則決策 intent thought.next_action_intent.lower() if “time” in intent: return Action(tool_name“get_current_time”, parameters{“timezone”: “Asia/Shanghai”}) elif “calculate” in intent or “math” in intent: # 這里需要從thought.known_info或更復(fù)雜的解析中獲取表達(dá)式 # 僅為示例 return Action(tool_name“calculator”, parameters{“expression”: “3 5 * 2”}) else: # 無(wú)法決策返回一個(gè)請(qǐng)求幫助的Action或者調(diào)用一個(gè)“clarify”工具 # 更合理的做法是這里集成LLM調(diào)用 # 模擬LLM返回一個(gè)錯(cuò)誤 return None async def _update_thought(self, old_thought: Thought, action: Action, observation: Observation) - Optional[Thought]: 更新?tīng)顟B(tài)邏輯。這里同樣需要集成LLM。 # 簡(jiǎn)化將結(jié)果存入known_info并設(shè)置一個(gè)簡(jiǎn)單的下一步意圖 new_thought old_thought.model_copy() new_thought.known_info[f“step_{old_thought.step}_result”] observation.data if action.tool_name “get_current_time”: new_thought.next_action_intent “時(shí)間已獲取任務(wù)完成” new_thought.goal_achieved True # 假設(shè)獲取時(shí)間就是目標(biāo) elif action.tool_name “calculator”: new_thought.next_action_intent “計(jì)算完成任務(wù)結(jié)束” new_thought.goal_achieved True else: new_thought.next_action_intent “分析上一步結(jié)果并決定下一步” return new_thought async def _handle_error(self, thought: Thought, failed_action: Action, error_obs: Observation) - bool: 錯(cuò)誤處理策略。返回True表示錯(cuò)誤已處理可繼續(xù)False表示致命錯(cuò)誤應(yīng)停止。 # 簡(jiǎn)單策略如果是calculator工具的參數(shù)錯(cuò)誤嘗試修正一次示例 if failed_action.tool_name “calculator” and “invalid characters” in error_obs.message: print(f“檢測(cè)到計(jì)算表達(dá)式錯(cuò)誤嘗試修正...”) # 這里可以嘗試清理表達(dá)式或使用備用表達(dá)式 # 本例中我們只是記錄并繼續(xù)實(shí)際應(yīng)更新thought和action # 例如將known_info中的某個(gè)標(biāo)記設(shè)為True讓_decide_action下次選擇其他路徑 thought.known_info[“calculator_failed”] True return True # 錯(cuò)誤已處理可以繼續(xù)下一輪循環(huán)但本輪循環(huán)的action已經(jīng)失敗thought未更新下次循環(huán)會(huì)基于舊的thought決策這可能有問(wèn)題 # 對(duì)于其他錯(cuò)誤我們暫時(shí)無(wú)法處理返回False讓Agent停止 return False5.4 運(yùn)行示例與調(diào)試# 模擬一個(gè)LLM客戶端實(shí)際應(yīng)替換為OpenAI、文心一言等API調(diào)用 class MockLLMClient: async def generate_thought(self, history): # 模擬LLM生成Thought這里返回固定值 return Thought(...) async def generate_action(self, thought): # 模擬LLM生成Action return Action(...) # 初始化并運(yùn)行 async def main(): llm MockLLMClient() agent SimpleAgent(tool_registry, llm) result await agent.run(“幫我計(jì)算一下3加5乘以2等于多少”) print(f“\n最終結(jié)果: {result}”) # 輸出歷史記錄用于調(diào)試 print(“\n 執(zhí)行歷史 ) for step in agent.history: print(f“Step {step.step_id}:”) print(f“ Thought: {step.thought.next_action_intent}”) if step.action: print(f“ Action: {step.action.tool_name}”) if step.observation: print(f“ Observation: {step.observation.status}”) # 運(yùn)行 import asyncio asyncio.run(main())這個(gè)簡(jiǎn)易框架清晰地展示了一個(gè)Agent如何圍繞“三元組”進(jìn)行狀態(tài)流轉(zhuǎn)如何通過(guò)工具管理器調(diào)用外部能力以及如何在核心循環(huán)中嵌入錯(cuò)誤處理邏輯。你可以看到_handle_error函數(shù)就是你的熔斷、降級(jí)和重試策略的入口點(diǎn)。而tool_registry.execute方法則是工具調(diào)用的統(tǒng)一安全門。6. 從“能用”到“好用”高級(jí)模式與工程化考量當(dāng)我們搭建起一個(gè)基礎(chǔ)可運(yùn)行的Agent后下一步就是考慮如何讓它更可靠、更高效、更易于管理和擴(kuò)展。這完全就是后端系統(tǒng)從“Demo”到“生產(chǎn)級(jí)”的演進(jìn)過(guò)程。6.1 支持復(fù)雜任務(wù)的工作流與子任務(wù)分解簡(jiǎn)單的線性“思考-行動(dòng)”循環(huán)對(duì)于復(fù)雜任務(wù)是不夠的。一個(gè)“規(guī)劃國(guó)慶旅行”的Agent需要先查天氣、再訂機(jī)票、然后訂酒店可能還需要查詢當(dāng)?shù)毓ヂ?。這些步驟之間有依賴關(guān)系訂酒店最好在確定目的地和日期之后也可能并行執(zhí)行查天氣和查攻略可以同時(shí)進(jìn)行。解決方案在Thought中引入任務(wù)棧或工作流引擎任務(wù)棧Thought中可以維護(hù)一個(gè)task_stack。當(dāng)主任務(wù)被分解為子任務(wù)時(shí)將子任務(wù)壓棧。當(dāng)前任務(wù)完成后再?gòu)棾鱿乱粋€(gè)任務(wù)。這類似于程序執(zhí)行的調(diào)用棧。集成工作流引擎對(duì)于極其復(fù)雜的場(chǎng)景可以直接將Agent作為工作流引擎的一個(gè)“智能節(jié)點(diǎn)”。工作流引擎如Airflow、Temporal負(fù)責(zé)定義任務(wù)DAG、調(diào)度和重試而Agent節(jié)點(diǎn)負(fù)責(zé)執(zhí)行其中需要“智能”的步驟如“根據(jù)用戶偏好從10個(gè)結(jié)果中推薦3個(gè)酒店”。這樣持久化、狀態(tài)恢復(fù)、并行化等難題就交給了成熟的引擎來(lái)處理。6.2 記憶與上下文管理避免“金魚腦”大模型有上下文長(zhǎng)度限制Agent也不能無(wú)限記住所有歷史。如何讓Agent擁有“長(zhǎng)期記憶”和“短期工作記憶”向量數(shù)據(jù)庫(kù)作為長(zhǎng)期記憶這是目前的主流方案。將重要的歷史交互如用戶偏好、已執(zhí)行的成功操作、學(xué)到的知識(shí)轉(zhuǎn)換成向量存入像Pinecone、Chroma這樣的向量數(shù)據(jù)庫(kù)。當(dāng)需要相關(guān)信息時(shí)通過(guò)向量相似度檢索召回。這相當(dāng)于后端系統(tǒng)的“緩存”或“歸檔數(shù)據(jù)庫(kù)”。Thought的摘要與提煉不要將原始的、冗長(zhǎng)的Observation全部塞進(jìn)下一次的Thought。而是設(shè)計(jì)一個(gè)“摘要”工具或步驟將冗長(zhǎng)的結(jié)果提煉成關(guān)鍵信息。例如搜索返回了10條航班信息可以摘要為“找到10個(gè)航班最早一班是XX最便宜一班是YY價(jià)格范圍是ZZ?！?這大大節(jié)省了上下文空間也提高了思考效率。分層記憶結(jié)構(gòu)可以設(shè)計(jì)會(huì)話級(jí)記憶本次對(duì)話、任務(wù)級(jí)記憶當(dāng)前復(fù)雜任務(wù)、用戶級(jí)記憶用戶長(zhǎng)期偏好。不同級(jí)別的記憶有不同的存儲(chǔ)和檢索策略。6.3 可觀測(cè)性與調(diào)試給Agent裝上“黑匣子”一個(gè)行為不可預(yù)測(cè)的Agent是可怕的。我們必須能監(jiān)控、記錄和調(diào)試它。結(jié)構(gòu)化日志Agent的每一個(gè)步驟三元組都必須被完整、結(jié)構(gòu)化地記錄下來(lái)。日志應(yīng)包含時(shí)間戳、會(huì)話ID、步驟ID、完整的Thought/ Action/ Observation數(shù)據(jù)、耗時(shí)、錯(cuò)誤信息等。這比打印文本日志強(qiáng)大得多便于后續(xù)分析和重放。追蹤與鏈路為每個(gè)用戶會(huì)話或任務(wù)生成唯一的trace_id并貫穿所有的工具調(diào)用。這樣當(dāng)工具調(diào)用失敗時(shí)你可以通過(guò)trace_id在后端各個(gè)微服務(wù)的日志中串聯(lián)起完整的調(diào)用鏈快速定位問(wèn)題。可視化調(diào)試界面開(kāi)發(fā)一個(gè)簡(jiǎn)單的Web界面能夠?qū)崟r(shí)查看Agent的執(zhí)行歷史以流程圖或時(shí)間線形式并能手動(dòng)編輯或回退到某一步的Thought重新執(zhí)行。這對(duì)于開(kāi)發(fā)和排查復(fù)雜問(wèn)題至關(guān)重要。6.4 性能與成本優(yōu)化Agent的思考調(diào)用大模型和工具調(diào)用調(diào)用API都可能很昂貴、很慢。緩存對(duì)于確定性較高的工具調(diào)用如查詢某個(gè)靜態(tài)信息、計(jì)算固定表達(dá)式結(jié)果可以緩存。對(duì)于LLM的思考如果相同的Thought在相同上下文中總是產(chǎn)生相同的Action也可以考慮緩存。但要注意緩存的有效性和安全性。異步與并行如果多個(gè)工具調(diào)用之間沒(méi)有依賴關(guān)系一定要并行執(zhí)行。asyncio.gather()是你的好朋友。這能顯著降低任務(wù)的總耗時(shí)。LLM調(diào)用優(yōu)化思維鏈CoT壓縮在將歷史對(duì)話喂給LLM生成下一步Thought時(shí)可以對(duì)之前的CoT進(jìn)行壓縮或總結(jié)只保留精華。模型分級(jí)不是每一步思考都需要用最強(qiáng)大、最貴的模型。可以設(shè)計(jì)一個(gè)路由策略簡(jiǎn)單決策用小模型復(fù)雜規(guī)劃再用大模型。提前停止如果Agent已經(jīng)明確達(dá)到了終止條件如成功、失敗、需要人工介入就不要再調(diào)用LLM做無(wú)謂的“下一步思考”了。7. 總結(jié)與個(gè)人體會(huì)走完這一趟從后端視角拆解Agent的旅程我的最大感受是Agent開(kāi)發(fā)本質(zhì)上是“智能”與傳統(tǒng)“軟件工程”的深度融合。大語(yǔ)言模型提供了前所未有的自然語(yǔ)言理解和生成能力但它不穩(wěn)定、不可控、成本高。而軟件工程尤其是后端分布式系統(tǒng)領(lǐng)域積累的幾十年經(jīng)驗(yàn)——模塊化、接口標(biāo)準(zhǔn)化、狀態(tài)管理、錯(cuò)誤處理、可觀測(cè)性——正是用來(lái)約束和賦能這種“智能”使其變得可靠、可用、可維護(hù)的關(guān)鍵。不要被“智能體”、“自主”這些詞嚇到。當(dāng)你開(kāi)始動(dòng)手時(shí)請(qǐng)時(shí)刻記住狀態(tài)是核心用“三元組”清晰地定義和記錄狀態(tài)這是所有邏輯的基礎(chǔ)。工具是手腳像設(shè)計(jì)微服務(wù)API一樣設(shè)計(jì)你的工具明確、健壯、可復(fù)用。錯(cuò)誤是常態(tài)假設(shè)一切都會(huì)出錯(cuò)并為此設(shè)計(jì)層層防線。重試、降級(jí)、熔斷、超時(shí)這些分布式系統(tǒng)的老套路在Agent世界里一樣好使。可觀測(cè)性是生命線沒(méi)有日志、沒(méi)有追蹤、無(wú)法重現(xiàn)問(wèn)題的Agent就像在黑暗中閉眼開(kāi)車。我自己的幾個(gè)項(xiàng)目從最初的“一把梭”調(diào)用GPT到現(xiàn)在擁有完整的狀態(tài)機(jī)、工具庫(kù)、錯(cuò)誤恢復(fù)和監(jiān)控面板穩(wěn)定性和用戶體驗(yàn)有了質(zhì)的提升。這條路沒(méi)有捷徑就是把這些后端工程的基本功扎實(shí)地應(yīng)用到Agent的架構(gòu)中去。希望這篇基于后端思維的長(zhǎng)文能給你帶來(lái)一些不一樣的、更落地的啟發(fā)。下次當(dāng)你再看到那些炫酷的Agent演示時(shí)不妨想想它的“三元組”在哪里流轉(zhuǎn)它的“工具調(diào)用”失敗了會(huì)怎樣它的“錯(cuò)誤處理”防線有幾層。想明白了這些你離自己構(gòu)建一個(gè)生產(chǎn)可用的Agent就不遠(yuǎn)了。