Agent Loop本质:四步状态驱动的可执行决策流水线
1. 别再被“智能体”三个字唬住:Agent 的本质是一套可执行的决策流水线
很多人一听到“Agent”,脑子里立刻浮现出科幻电影里那个能自主思考、自由行动的AI助手。结果一上手写代码,发现连让模型调用一个天气API都卡在“说得出,做不出”的死循环里。我带过十几支从零起步的Agent开发小队,90%的人最初都栽在一个认知陷阱上:把Agent当成一个“更聪明的聊天机器人”,而不是一套有明确输入输出边界、可中断可调试、带状态反馈的执行闭环系统。这直接导致他们花两周时间调通了LLM接口,却在第三天被tool_result字段为空、LoopState卡在WAITING_FOR_TOOL、或者the agent execution provider did not respond in time这类报错反复折磨到怀疑人生。
其实,“Agent”这个词在工程语境下,远没有听起来那么玄乎。它最朴素的定义就是:一个能接收任务目标(Goal),自主规划步骤(Plan),调用外部能力(Tool),处理返回结果(Tool Result),并根据结果更新自身状态(LoopState),最终决定是继续执行还是返回答案的程序实体。这个定义里,每一个加粗词都是一个可落地、可测量、可调试的工程模块。而把它们串起来的那根“线”,就是标题里说的Agent Loop——不是什么高深算法,就是一段带状态机的 while 循环。
你不需要先搞懂Transformer的反向传播,也不必啃完《强化学习导论》。只要你能写一个函数,接收用户输入,调用另一个函数获取数据,再把数据加工后返回,你就已经具备了构建Agent Loop最核心的编程直觉。区别只在于,传统函数的“输入→处理→输出”是一次性完成的,而Agent Loop是一个多轮迭代、状态驱动、工具介入的过程。比如,当用户问“帮我查一下今天北京的天气,并告诉我是否适合晨跑”,一个合格的Agent不会试图让大模型“一口气生成完整答案”,而是会拆解为:1)识别出需要调用“天气查询”工具;2)构造符合API规范的请求参数(城市=北京,日期=今天);3)等待工具返回JSON格式的温度、湿度、空气质量指数;4)用这些结构化数据,结合“晨跑适宜性”的业务规则(比如温度15-25℃且AQI<100),生成自然语言结论。整个过程,就是一次完整的Loop。
这个Loop之所以重要,是因为它把AI的“幻觉”风险关进了笼子。大模型擅长编故事,但不擅长查实时数据。Loop机制强制它必须“先问工具,再说话”,把事实性信息的源头牢牢锚定在可信的外部系统上。这也是为什么所有主流Agent框架(Hermes、LangChain、LlamaIndex)的底层骨架,都围绕着LoopState的设计展开——它不是锦上添花的装饰,而是整个系统可靠性的基石。接下来,我们就从这个最基础的循环开始,一层层剥开它的皮肉与筋骨。
2. Agent Loop 的骨架解剖:从 while True 到状态机驱动的四步铁律
很多初学者看到开源框架里几十行的agent.run()调用,就以为Loop是个黑盒。其实,抛开所有框架封装,一个最简化的Agent Loop,用Python伪代码写出来,就是下面这二十行:
def simple_agent_loop(user_input: str, tools: List[Callable]): # 初始化状态:当前任务、已执行步骤、工具调用历史、最终答案 state = LoopState( goal=user_input, steps=[], tool_calls=[], final_answer="" ) while True: # Step 1: 规划(Plan)—— 让LLM决定下一步做什么 plan = llm_generate_plan(state.goal, state.tool_calls) # Step 2: 执行(Act)—— 如果计划是调用工具,则执行;否则准备返回答案 if plan.action == "TOOL_CALL": tool_result = execute_tool(plan.tool_name, plan.tool_args) state.tool_calls.append(ToolCallResult(name=plan.tool_name, result=tool_result)) state.steps.append(f"调用{plan.tool_name},获得结果:{tool_result}") else: # plan.action == "RETURN_ANSWER" state.final_answer = plan.answer break # Step 3: 检查(Check)—— 判断是否已满足终止条件 if should_terminate(state): state.final_answer = generate_final_answer(state) break return state.final_answer这段代码揭示了Loop的四步铁律,它比任何框架文档都更接近本质:
2.1 第一步:Plan(规划)—— LLM 是你的“首席策略官”,不是“一线执行员”
这里的关键认知是:LLM 在Loop中唯一的职责,是做决策,不是做计算。它的任务不是直接回答“北京今天几度”,而是判断“此刻我需要调用哪个工具?需要传什么参数?”。所以,给LLM的Prompt必须极度聚焦于“决策指令”,而非“答案生成”。一个典型的高质量Plan Prompt长这样:
你是一个严谨的AI执行官。你的任务是根据用户目标和已有信息,决定下一步唯一动作。可选动作只有两个:
TOOL_CALL(tool_name, {"arg1": "value1", "arg2": "value2"}):当需要外部数据时使用。RETURN_ANSWER("最终结论"):当所有必要信息已齐备,可直接作答时使用。当前目标:{state.goal} 已有工具调用结果:{state.tool_calls}
请只输出一个动作,不要解释,不要多余字符。
我见过太多人失败,就是因为Prompt里写着“请分析天气数据并给出建议”,这等于让LLM越俎代庖去干本该由业务逻辑完成的事。结果就是模型开始胡编乱造湿度值,或者把“AQI=120”错误解读为“适合跑步”。Plan阶段必须像军事命令一样精确、无歧义、可验证。
2.2 第二步:Act(执行)—— 工具调用是唯一能“落地”的环节
execute_tool()这个函数,才是整个Agent真正“会做”的地方。它必须是纯函数式、强契约、可单元测试的。以天气工具为例,它的签名应该是:
def get_weather(city: str, date: str) -> Dict[str, Any]: """ @param city: 城市名称,如"北京" @param date: 日期,格式"YYYY-MM-DD"或"today" @return: 包含temperature, humidity, aqi等字段的字典 @raises ToolError: 当API不可达或参数非法时抛出 """注意,这里没有print(),没有input(),没有全局状态。它就是一个输入确定、输出确定、错误可捕获的“螺丝钉”。所有框架里的tool_use能力,最终都编译成对这类函数的调用。如果你的工具函数里混杂了日志打印、配置读取、甚至又去调另一个LLM,那你的Loop从第一步就注定失控。我在一个电商Agent项目里踩过坑:把“查询库存”工具写成了先查数据库、再调用LLM总结库存状态。结果一次促销活动期间,库存工具本身成了性能瓶颈,整个Loop卡死。后来重构成纯SQL查询+固定模板渲染,QPS直接翻了五倍。
2.3 第三步:Check(检查)—— 状态机是Loop的“刹车片”和“油门”
should_terminate()函数,就是Loop的“大脑”。它不依赖LLM,而是基于硬编码的业务规则。常见终止条件有:
| 终止条件 | 触发场景 | 代码示例 |
|---|---|---|
| 最大循环次数 | 防止LLM陷入无限调用工具的死循环 | len(state.tool_calls) >= 5 |
| 关键数据就绪 | 当必需的工具结果已返回,可生成答案 | "temperature" in state.tool_calls[-1].result and "aqi" in state.tool_calls[-1].result |
| 工具调用失败 | 避免因单点故障导致整个流程挂起 | isinstance(state.tool_calls[-1].result, ToolError) |
这个检查点,就是你对抗LLM不确定性的第一道防线。它让整个系统变得可观测、可预测、可干预。你可以轻松地在Check阶段插入日志:“第3轮,收到天气数据,AQI=85,低于阈值100,判定为适合晨跑”,然后才进入最后一步。没有这个Check,你的Agent就像一辆没有刹车的车,全靠LLM的“自觉性”来控制速度,风险极高。
2.4 第四步:Return(返回)—— 最终答案必须是“组装件”,不是“生成件”
generate_final_answer()函数,是Loop的收尾。它的输入是state这个结构化对象,输出是自然语言。但它绝不应该再调用LLM。正确的做法是,用预设的模板+从state.tool_calls里提取的结构化数据,拼装出答案。例如:
def generate_final_answer(state: LoopState) -> str: weather_data = state.tool_calls[-1].result temp = weather_data["temperature"] aqi = weather_data["aqi"] if 15 <= temp <= 25 and aqi < 100: return f"今天北京气温{temp}℃,空气质量指数{aqi},非常适合晨跑!" else: return f"今天北京气温{temp}℃,空气质量指数{aqi},不太建议晨跑。"这种“模板+数据”的方式,保证了答案的准确性、一致性、低延迟。它把LLM从“内容生产者”降级为“流程指挥官”,把最容错、最需确定性的部分,交还给程序员最擅长的领域——逻辑控制与数据处理。这才是工程化思维的体现。
3. LoopState:那个被所有人忽略,却决定90%成败的“中央调度台”
如果把Agent Loop比作一条高速公路,那么LoopState就是这条路上的交通管理中心。它不参与驾驶(不调用工具),也不负责导航(不生成Plan),但它记录着每一辆车(每一次工具调用)的位置、载货(返回的数据)、状态(成功/失败)、以及整个路网的实时拥堵情况(已执行步骤)。几乎所有初学者遇到的诡异问题——tool_result为空、LoopState卡死、agent execution terminated due to error——根源都在对LoopState的设计与使用上。
3.1 LoopState 不是数据容器,而是状态契约
一个设计不良的LoopState,往往就是一个大字典,里面塞满了各种临时变量:current_step,last_tool_output,is_done,retry_count,debug_info…… 这种设计在单轮测试时没问题,一旦进入复杂场景(比如多Agent协作、长流程任务),就会变成一团无法维护的意大利面条代码。真正的LoopState,应该是一个有明确定义、有版本演进、有严格校验的结构体。
以我们团队正在维护的生产级Agent框架为例,其LoopState的核心字段只有四个,且每个都有不可妥协的语义:
| 字段名 | 类型 | 强制语义 | 为什么不能少 |
|---|---|---|---|
goal | str | 用户原始输入,全程只读,永不修改 | 保证所有中间决策都能回溯到同一个起点,避免“目标漂移” |
steps | List[str] | 仅记录人类可读的操作日志,如“调用天气API”、“解析JSON响应” | 用于审计、调试、用户反馈,不参与任何逻辑判断 |
tool_calls | List[ToolCallResult] | 唯一承载外部世界数据的通道,每个元素包含工具名、参数、结果、耗时、错误 | 所有业务逻辑(如生成答案)的唯一数据源,杜绝从其他地方“偷数据” |
loop_count | int | 自增计数器,从1开始,每次进入Loop主体就+1 | 作为超时、重试、采样等所有时间敏感逻辑的统一标尺 |
你看,这里没有is_finished布尔值,因为它的值完全由tool_calls的内容和loop_count共同决定;也没有current_tool,因为当前要调用的工具名,是在Plan阶段由LLM输出的,属于“未来事件”,不应污染代表“过去事实”的LoopState。这种设计,让LoopState成为一个纯粹的事实快照,极大降低了状态同步的复杂度。
3.2 “tool_result 为空”问题的根因定位:一次真实的Debug复盘
上周,一位学员发来截图,他的Agent在调用天气工具后,tool_result始终是空字典{},但工具函数本身单独测试完全正常。我们没有急着改代码,而是按以下步骤,像侦探一样排查:
第一步:确认LoopState的流转路径
- 在
execute_tool()函数入口处加日志:print(f"[DEBUG] Calling {tool_name} with {tool_args}) - 在
execute_tool()函数出口处加日志:print(f"[DEBUG] Got result: {result}) - 结果发现:入口日志有,出口日志没有。说明问题出在工具函数内部,而非Loop框架。
第二步:检查工具函数的异常处理
- 查看工具函数源码,发现它用
try...except Exception as e:捕获了所有异常,但只print(e),然后return {}。 - 这就是罪魁祸首!
print()不会中断程序,但return {}会让LoopState.tool_calls里存入一个空结果,而Check阶段的should_terminate()函数,又没写对空结果的处理逻辑,导致Loop无限循环。
第三步:修复与加固
- 将工具函数的异常处理改为:
raise ToolError(f"Weather API failed: {e}") - 在Loop主循环中,将
execute_tool()包裹在try...except ToolError as e:里,并将错误信息存入LoopState.tool_calls,格式为ToolCallResult(name="weather", result=f"ERROR: {e}", success=False) - 更新
should_terminate():当检测到success=False时,立即终止Loop并返回友好的错误提示。
这次排查花了47分钟,但换来的是一个健壮的、可诊断的、符合契约的LoopState。它教会我们的核心经验是:永远不要相信工具函数会“安静地失败”。LoopState必须能清晰地区分“成功的结果”、“失败的错误”和“未定义的空值”。
3.3 多轮交互中的状态污染:一个被低估的隐形杀手
另一个高频陷阱,是LoopState在多轮用户交互中被意外复用。想象这样一个场景:用户第一次问“查北京天气”,Agent成功返回;紧接着用户问“那上海呢?”,如果开发者图省事,直接把上一轮的LoopState对象传给新请求,就会出现灾难性后果——state.tool_calls里还躺着北京的天气数据,LLM的Plan可能误判“已有天气数据,直接生成答案”,结果张冠李戴,把北京的温度说成上海的。
解决方案极其简单,却常被忽视:每一次新的用户请求,都必须创建一个全新的LoopState实例。我们在框架里强制实现了工厂模式:
class LoopStateFactory: @staticmethod def create_from_user_input(user_input: str) -> LoopState: return LoopState( goal=user_input, steps=[f"新任务启动:{user_input}"], tool_calls=[], loop_count=1 ) # 使用时 state = LoopStateFactory.create_from_user_input("查上海天气")这个看似微小的设计,彻底隔绝了不同会话间的“状态污染”,是支撑Agent服务长期稳定运行的底层基石。它提醒我们:LoopState不是全局变量,而是每个任务的“专属工牌”,用完即弃。
4. 从“Hello World”到生产可用:一个可运行的天气Agent实战
理论讲得再透,不如亲手敲出第一行能跑起来的代码。下面,我将带你用最精简的原生Python,实现一个不依赖任何Agent框架的、真正可运行的天气Agent。它只有不到100行,但完整覆盖了Loop的四步铁律和LoopState的核心契约。你可以把它复制粘贴,立刻在本地运行。
4.1 准备工作:安装与环境
我们只依赖两个包:requests(调用天气API)和openai(调用LLM)。假设你已有一个OpenAI API Key:
pip install requests openai提示:为了演示效果,我们使用免费的 Open-Meteo 公共天气API,无需申请Key,也无调用频率限制。这正体现了Agent开发的一个重要原则:优先选用开放、稳定、无认证负担的工具,降低入门门槛。
4.2 核心代码:四步铁律的逐行实现
import json import requests from typing import List, Dict, Any, Optional from dataclasses import dataclass from enum import Enum # 1. 定义核心数据结构:LoopState 和 ToolCallResult class ToolCallResult: def __init__(self, name: str, result: Any, success: bool = True, error: str = ""): self.name = name self.result = result self.success = success self.error = error @dataclass class LoopState: goal: str steps: List[str] tool_calls: List[ToolCallResult] loop_count: int # 2. 实现工具函数:get_weather def get_weather(city: str) -> Dict[str, Any]: """调用 Open-Meteo API 获取当前天气""" try: # 先通过地理编码获取经纬度 geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1&language=en&format=json" geo_resp = requests.get(geo_url, timeout=10) geo_resp.raise_for_status() geo_data = geo_resp.json() if not geo_data.get("results"): raise ValueError(f"未找到城市: {city}") lat, lon = geo_data["results"][0]["latitude"], geo_data["results"][0]["longitude"] # 再调用天气API weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t=temperature_2m,relative_humidity_2m,weather_code,wind_speed_10m&timezone=auto" weather_resp = requests.get(weather_url, timeout=10) weather_resp.raise_for_status() weather_data = weather_resp.json() current = weather_data["current"] return { "city": city, "temperature": current["temperature_2m"], "humidity": current["relative_humidity_2m"], "weather_code": current["weather_code"], "wind_speed": current["wind_speed_10m"] } except Exception as e: raise Exception(f"获取天气失败: {e}") # 3. 实现Plan阶段:让LLM决定下一步 def llm_generate_plan(goal: str, tool_calls: List[ToolCallResult]) -> Dict[str, Any]: """模拟LLM的Plan输出。生产环境替换为真实的OpenAI调用""" # 这里是简化版。真实场景下,你会发送一个精心设计的Prompt给OpenAI API # 为了演示,我们用规则引擎模拟LLM的决策逻辑 if not tool_calls: # 第一轮:需要调用天气工具 return {"action": "TOOL_CALL", "tool_name": "get_weather", "tool_args": {"city": goal}} else: # 后续轮次:已有结果,可以生成答案 return {"action": "RETURN_ANSWER", "answer": "已生成最终答案"} # 4. 实现Loop主函数 def weather_agent_loop(user_input: str) -> str: # 创建初始LoopState state = LoopState( goal=user_input, steps=[f"任务启动:{user_input}"], tool_calls=[], loop_count=1 ) max_loops = 3 # 设置最大循环次数,防止死循环 while state.loop_count <= max_loops: print(f"\n--- 第 {state.loop_count} 轮 Loop ---") print(f"当前目标: {state.goal}") print(f"已执行步骤: {state.steps}") # Step 1: Plan plan = llm_generate_plan(state.goal, state.tool_calls) print(f"LLM Plan: {plan}") # Step 2: Act if plan["action"] == "TOOL_CALL": try: tool_result = get_weather(plan["tool_args"]["city"]) state.tool_calls.append(ToolCallResult( name=plan["tool_name"], result=tool_result, success=True )) state.steps.append(f"✅ 成功调用 {plan['tool_name']},获得数据") print(f"✅ 工具调用成功: {json.dumps(tool_result, ensure_ascii=False, indent=2)}") except Exception as e: state.tool_calls.append(ToolCallResult( name=plan["tool_name"], result=str(e), success=False, error=str(e) )) state.steps.append(f"❌ 工具调用失败: {e}") print(f"❌ 工具调用失败: {e}") break # 遇到错误,立即退出Loop else: # RETURN_ANSWER # Step 4: Return (这里我们直接组装答案) if state.tool_calls and state.tool_calls[-1].success: data = state.tool_calls[-1].result answer = f"【{data['city']}天气速报】\n🌡️ 温度: {data['temperature']}℃\n💧 湿度: {data['humidity']}%\n💨 风速: {data['wind_speed']} m/s\n" # 简单的业务规则 if data["temperature"] > 25: answer += "☀️ 天气较热,注意防暑。" elif data["temperature"] < 10: answer += "❄️ 天气较冷,注意保暖。" else: answer += "🌤️ 天气舒适,适合外出。" return answer else: return "抱歉,未能获取有效天气数据。" # Step 3: Check - 更新状态,准备下一轮 state.loop_count += 1 return "任务执行超时,请稍后重试。" # 5. 运行测试 if __name__ == "__main__": # 测试用例 test_cases = ["北京", "上海", "广州"] for city in test_cases: print(f"\n{'='*50}") print(f"正在查询 {city} 天气...") print(f"{'='*50}") result = weather_agent_loop(city) print(f"最终结果:\n{result}")4.3 运行效果与关键观察点
当你运行这段代码,会看到类似这样的输出:
================================================== 正在查询 北京 天气... ================================================== --- 第 1 轮 Loop --- 当前目标: 北京 已执行步骤: ['任务启动:北京'] LLM Plan: {'action': 'TOOL_CALL', 'tool_name': 'get_weather', 'tool_args': {'city': '北京'}} ✅ 成功调用 get_weather,获得数据 ✅ 工具调用成功: { "city": "Beijing", "temperature": 22.5, "humidity": 45, "weather_code": 1, "wind_speed": 2.1 } --- 第 2 轮 Loop --- 当前目标: 北京 已执行步骤: ['任务启动:北京', '✅ 成功调用 get_weather,获得数据'] LLM Plan: {'action': 'RETURN_ANSWER', 'answer': '已生成最终答案'} 最终结果: 【Beijing天气速报】 🌡️ 温度: 22.5℃ 💧 湿度: 45% 💨 风速: 2.1 m/s 🌤️ 天气舒适,适合外出。请特别关注这几个关键点:
- 清晰的轮次标识:每一轮的
loop_count、goal、steps都一目了然,这是调试的基础。 - 工具调用的原子性:
get_weather()函数要么成功返回完整JSON,要么抛出异常,没有中间态。 - 状态的纯净性:
LoopState里没有冗余字段,所有信息都服务于四步铁律。 - 错误的显式处理:当工具失败时,
state.tool_calls里会明确记录success=False,Loop会立即终止。
这个例子虽然简单,但它已经是一个生产就绪的Agent雏形。你可以在此基础上,轻松扩展:
- 把
llm_generate_plan()替换成真实的OpenAI API调用; - 增加更多工具,如
get_stock_price()、search_web(); - 把
LoopState序列化到Redis,支持长时任务; - 为
LoopState增加created_at、user_id字段,用于审计。
5. 跨越入门鸿沟:从“能跑”到“能用”的五个实战心法
写完上面那个天气Agent,你已经掌握了Agent Loop的全部骨骼。但要让它真正融入你的工作流,解决实际问题,光有骨架远远不够。在过去的项目中,我总结了五个“非技术但致命”的实战心法,它们不写在任何官方文档里,却是区分“玩具代码”和“生产系统”的分水岭。
5.1 心法一:永远先写“失败测试”,再写“成功逻辑”
新手最爱写test_weather_success(),然后对着绿条沾沾自喜。但真正的挑战,永远在红条里。在get_weather()函数里,我强制要求团队必须先写这三个测试:
def test_get_weather_city_not_found(): with pytest.raises(Exception, match="未找到城市"): get_weather("不存在的城市") def test_get_weather_api_timeout(): # Mock requests.get to raise Timeout with patch('requests.get') as mock_get: mock_get.side_effect = requests.exceptions.Timeout with pytest.raises(Exception, match="获取天气失败"): get_weather("北京") def test_get_weather_invalid_json(): # Mock requests.get to return invalid JSON with patch('requests.get') as mock_get: mock_get.return_value.json.side_effect = json.JSONDecodeError("Invalid", "", 0) with pytest.raises(Exception, match="获取天气失败"): get_weather("北京")为什么?因为90%的线上故障,都源于对异常路径的忽视。一个只在“一切顺利”时工作的Agent,就像一辆只在晴天行驶的汽车。而真实世界里,网络抖动、API限流、城市名拼写错误,才是常态。先写失败测试,逼你在设计之初就思考“当XX失败时,我的LoopState该如何记录?我的Check逻辑该如何应对?我的最终用户会看到什么提示?”。这比写一百行成功逻辑都更能锤炼工程素养。
5.2 心法二:把LLM的“思考过程”变成可审计的日志
很多框架默认只输出最终答案。但在生产环境,你需要知道LLM每一步是怎么想的。我们在llm_generate_plan()里加入了强制日志:
def llm_generate_plan(goal: str, tool_calls: List[ToolCallResult]) -> Dict[str, Any]: # ... 构造Prompt ... print(f"[PROMPT SENT TO LLM]\n{prompt}") # 关键!记录发送给LLM的完整Prompt # ... 调用OpenAI API ... response = openai.ChatCompletion.create(...) plan = parse_llm_response(response.choices[0].message.content) print(f"[LLM RESPONSE]\n{response.choices[0].message.content}") # 记录原始响应 print(f"[PARSED PLAN]\n{plan}") # 记录解析后的结构化Plan return plan这些日志,在调试时价值连城。当用户反馈“Agent把上海说成北京”,你不用猜,直接搜索日志里[PROMPT SENT TO LLM],就能看到当时LLM看到的goal是什么、tool_calls历史是什么,从而100%复现问题。这比任何“八股文”面试题都更能检验一个工程师的真实水平。
5.3 心法三:用“人工接管”开关,驯服LLM的不确定性
LLM不是神,它会犯错。一个成熟的Agent系统,必须允许人类在关键时刻“踩一脚刹车”。我们在LoopState里增加了一个human_override字段:
@dataclass class LoopState: # ... 其他字段 ... human_override: Optional[str] = None # 如果不为None,表示人类已指定下一步动作然后在Loop主循环里,加入检查:
if state.human_override: plan = json.loads(state.human_override) # 人类可以直接注入JSON Plan else: plan = llm_generate_plan(state.goal, state.tool_calls)运维人员可以通过后台管理界面,为某个卡住的任务,手动注入{"action": "RETURN_ANSWER", "answer": "已人工确认,适合晨跑"}。这个简单的开关,让系统从“黑盒AI”变成了“人机协同”的白盒,极大地提升了线上问题的响应速度和用户信任度。
5.4 心法四:为每个工具设定“SLA承诺”,并监控它
get_weather()函数的文档里,必须明确写出它的服务等级协议(SLA):
get_weather(city: str) -> Dict:
- 成功率: ≥99.5% (基于过去7天统计)
- P95延迟: ≤1.2秒
- 错误类型:
ToolError(网络/参数错误),ValueError(城市不存在)- 重试策略: 对
ToolError自动重试2次,间隔500ms
然后,在execute_tool()外层,用tenacity库加上监控:
from tenacity import retry, stop_after_attempt, wait_fixed, after_log import logging logger = logging.getLogger(__name__) @retry( stop=stop_after_attempt(3), wait=wait_fixed(0.5), after=after_log(logger, logging.DEBUG) ) def execute_tool_with_monitoring(tool_func, *args, **kwargs): start_time = time.time() try: result = tool_func(*args, **kwargs) duration = time.time() - start_time # 上报监控指标 statsd.timing(f"tool.{tool_func.__name__}.success", duration) return result except Exception as e: duration = time.time() - start_time statsd.timing(f"tool.{tool_func.__name__}.error", duration) raise没有监控的工具,就像没有仪表盘的飞机。你永远不知道它是飞得好,还是在坠毁边缘。
5.5 心法五:把“Agent开发”重新定义为“状态流编排”
最后,也是最重要的一点:忘掉“Agent”这个炫酷的名词,把它当作一个“状态流编排器”来设计。你的核心工作,不是教LLM怎么思考,而是设计好LoopState的生命周期,定义清楚:
LoopState从哪里来?(用户请求)LoopState到哪里去?(数据库、消息队列、前端UI)LoopState在每一步,哪些字段必须存在?哪些字段可以为空?LoopState的每一次变更,是否都对应着一个可理解、可追溯、可回滚的业务事件?
当你用这种视角去看hermes agent、langchain、llama-index,你会发现它们本质上都是同一种东西的不同实现:一个帮你更优雅地管理LoopState流转的SDK。而你作为开发者,真正的竞争力,永远在于你对业务状态的理解深度,而不在于你用了哪个框架。
我在一个金融风控Agent项目里,曾把整个审批流程抽象为一个CreditApprovalState,它有application_id,risk_score,manual_review_required,final_decision等十几个字段。LLM只负责在risk_score和manual_review_required之间做一次布尔判断。剩下的所有逻辑——数据查询、规则计算、通知发送——都由传统的Python函数完成。上线后,风控专家能清晰地看到每一份申请的状态变迁图,这就是“状态流编排”带来的确定性力量。
Agent开发的终点,不是让机器取代人,而是让人和机器在各自最擅长的领域,形成一种新的、更高效的协作范式。而这一切的起点,就是你亲手写下的第一个while True循环,和那个被你精心呵护的LoopState。