1. 循环Loops入门指南从基础概念到智能体工作流实战最近在开发一个自动化数据处理工具时我又一次被“循环”这个概念给卡住了。不是简单的for或while而是需要让一个AI智能体根据中间结果动态决定下一步是继续分析、调用外部API还是生成最终报告。这种“会思考的循环”正是当前AI应用开发特别是智能体Agent领域的核心。无论你是想用Claude Code搭建一个自动代码审查机器人还是设计一个能自主完成多步骤任务的Goal-Based智能体不理解循环的深层机制就只能在门口打转。很多人对循环的理解还停留在编程课本里的迭代次数但现代AI开发中的循环尤其是Agentic Loops智能体循环和Turn-based Loops回合制循环其复杂度和威力远超传统认知。它不再是机械地重复而是包含了状态管理、条件评估、工具调用和策略选择的动态工作流引擎。今天我就结合自己踩过的坑和实战经验带你彻底搞懂循环并手把手展示如何用流行的工具比如Claude Code来构建真正智能的循环逻辑。2. 循环的演进从迭代到智能决策2.1 传统循环的局限与核心要素我们最早接触的循环无论是for i in range(10)还是while condition本质都是“预定轨道的重复”。它们有明确的边界一个起始点、一个终止条件、一个固定的步进动作。这种循环的“智能”程度为零它忠实地执行指令但对外界变化和自身产生的中间结果毫无感知。然而在自动化任务处理尤其是需要与语言模型LLM交互的场景中这种简单循环立刻显得力不从心。比如你想让AI总结一份长文档传统思路把文档切成10段用循环调用10次API把10个结果拼起来。实际问题第5段可能提到“详情见附录”但循环器不知道需要去额外获取附录内容总结到第8段时可能发现前面几段的总结有矛盾需要回溯调整但循环已经回不去了。这里就引出了智能循环必须处理的几个核心要素状态State循环当前进行到哪一步已经生成了哪些信息这些信息构成了循环的“记忆”。评估Evaluation当前状态是否达到了目标或者是否出现了需要特别处理的情况如错误、歧义、新信息决策Decision基于评估结果下一步应该做什么是继续原流程还是跳转到其他步骤或是终止执行Execution根据决策执行具体操作如调用一个工具函数、生成一段文本、查询数据库等。2.2 Agentic Loops赋予循环“智能体”思维Agentic Loop是当前AI应用开发的热门范式。你可以把它理解为一个内置了“思考-行动-观察”循环的自主智能体。它的典型流程不是线性的而是一个循环感知/规划智能体分析当前目标Goal和状态State规划下一步要执行的动作Action。例如“用户想订机票。当前状态是已知目的地和大致日期。下一步动作应该是‘查询航班信息’。”执行智能体执行规划好的动作通常是调用一个工具Tool。例如调用一个“航班搜索API”。观察智能体获取动作执行的结果Observation。例如API返回了未来三天所有航班列表和价格。评估与循环智能体观察结果并更新内部状态。然后重新评估“根据返回的航班信息我的目标订到性价比高的票完成了吗”如果没有则回到步骤1进行下一轮“规划”。新的规划可能是“筛选出下午起飞且价格低于1000元的航班”。这个循环会一直进行直到智能体评估认为目标已达成如成功生成订单号或遇到无法逾越的障碍如所有航班售罄或达到预设的最大循环次数。注意设计Agentic Loop时最关键的是定义清晰的“停止条件”。否则智能体可能陷入无限循环或者在“觉得差不多”但实际上并未完成任务时提前退出。通常需要结合目标达成度评估和最大步数限制。2.3 Turn-based Loops 与 Goal-based Loops 辨析这两个概念经常被混用但它们侧重点不同。Turn-based Loops回合制循环更强调交互的“轮次”和“回合”。常见于对话机器人、游戏AI或多轮表单填写。每一轮Turn通常包含用户输入 - 系统处理 - 系统输出。循环的驱动因素是“是否有下一轮用户输入”。它的状态管理侧重于对话历史决策逻辑在于如何根据最新输入和历史上下文生成合适的回应。类比就像下棋你走一步一个Turn我根据棋盘状态State走一步如此循环。Goal-based Loops目标驱动循环更强调最终目标的达成。循环的驱动因素是“当前状态与目标的差距”。智能体的一切行动都为了缩小这个差距。它可能包含多个复杂的子步骤这些步骤不一定是线性的。类比就像规划一次旅行Goal抵达某地。你的行动买票、去机场、登机都是由“抵达目的地”这个目标驱动的循环决策过程。过程中可能需要多个“回合”如与售票员沟通、通过安检但核心主线是目标。在实际开发中一个复杂的智能体往往融合了这两种循环。例如一个客服智能体其顶层是一个Goal-based Loop目标解决用户问题而在解决过程中与用户的每一轮对话都是一个Turn-based Loop。3. 构建智能循环的核心组件与设计模式理解了概念我们来看看如何动手搭建。一个健壮的智能循环系统通常由以下几个核心组件构成。3.1 状态管理循环的“记忆中枢”状态是循环的基石。糟糕的状态设计会导致信息丢失、逻辑混乱。状态设计要点结构化不要用一堆零散的变量。建议使用一个字典Python dict或一个Pydantic模型来集中管理。# 一个简单的任务处理状态示例 from pydantic import BaseModel from typing import List, Optional class TaskState(BaseModel): goal: str # 原始目标如“总结文档A” current_step: str # 当前步骤名如“正在提取摘要” completed_steps: List[str] [] # 已完成步骤 extracted_info: dict {} # 已提取的信息 errors: List[str] [] # 遇到的错误 iteration_count: int 0 # 循环次数 is_finished: bool False # 是否完成 final_result: Optional[str] None # 最终结果持久化考虑对于长耗时任务状态可能需要保存到数据库或文件以便中断后恢复。可以在每个循环迭代结束后序列化状态。只读与可写部分明确哪些状态信息是只读的背景如初始目标哪些是会在循环中被修改的如已提取信息。这有助于理清逻辑。实操心得在状态中增加一个history字段记录每一步的决策和结果。这在调试时是无价之宝你可以清晰地看到智能体“思考”的轨迹哪里走了弯路一目了然。3.2 决策引擎循环的“大脑”决策引擎根据当前状态决定下一步行动。实现方式有多种基于规则的引擎最简单直接。使用if...elif...else语句。def decide_next_action(state: TaskState) - str: if state.iteration_count 10: return force_terminate elif error in state.extracted_info: return handle_error elif len(state.extracted_info) 5: # 假设收集够5条信息就总结 return generate_summary else: return extract_next_item优点逻辑清晰易于调试。缺点规则复杂后难以维护缺乏灵活性。基于LLM的引擎将状态和可用工具描述给LLM让LLM生成下一步指令。这是Agentic Loop的核心。# 伪代码示例 def llm_decide(state, available_tools): prompt f 你是一个任务处理助手。当前状态{state}。 你可以使用的工具有{available_tools}。 请分析状态决定下一步应该调用哪个工具直接输出工具名或者任务是否完成输出“FINISH”。 只需输出工具名或“FINISH”。 decision call_llm(prompt) # 调用LLM API return decision.strip()优点极其灵活能处理复杂、未预见的场景。缺点成本高、速度慢、输出可能不稳定需要好的提示工程。混合引擎结合两者优势。用规则处理简单、明确的路径用LLM处理复杂、需要推理的决策。这是目前最实用的方案。3.3 工具Tools与执行器循环的“手脚”工具是智能体与外界交互的手段。一个工具就是一个可执行的函数比如“搜索网页”、“执行计算”、“读写文件”。设计工具的关键功能单一且明确一个工具只做一件事。例如search_web(query)就比search_and_summarize(query)要好。后者把搜索和总结耦合了不利于复用和决策。提供清晰的描述LLM决策引擎需要根据工具描述来决定使用哪个。描述应简洁说明工具的功能、输入和输出。tools [ { name: get_weather, description: 获取指定城市的当前天气情况。, parameters: {city: {type: string, description: 城市名}}, function: get_weather_api # 实际的后端函数 }, { name: calculate, description: 执行一个数学计算表达式。, parameters: {expression: {type: string, description: 数学表达式如 12*3}}, function: eval_expression # 注意实际使用中要对eval做安全限制 } ]执行器Executor负责调用工具函数并将结果格式化更新到状态中。它需要处理工具调用可能出现的异常。3.4 主流设计模式解析在实际项目中循环的实现通常遵循几种常见模式ReAct (Reason Act) 模式这是Agentic Loop的经典范式。智能体在每一步输出一个“思考Thought”和一个“行动Action”。执行行动后得到“观察Observation”然后进入下一轮思考。流程Thought - Action - Observation - Thought - ...优点将推理过程显式化易于理解和调试。缺点每次循环都需要LLM生成“思考”Token消耗较大。Plan-and-Execute 模式智能体先制定一个完整的计划一系列步骤然后按顺序执行这个计划。在执行每个步骤时可以再套用简单的循环或规则。流程Plan - Step1 - Step2 - ... - Finalize优点整体方向明确可能减少总的LLM调用次数。缺点计划可能不符合实际执行中遇到的情况缺乏动态调整能力。Reflection 模式在ReAct基础上增加一个“反思Reflection”阶段。在一系列行动后智能体停下来回顾历史评估进展修正策略然后再继续。流程... - Action - Observation - Reflection - New Thought - ...优点能纠正错误从失败中学习更适合复杂长程任务。缺点进一步增加了复杂度和计算成本。选择哪种模式取决于你的任务。简单、线性的任务适合Plan-and-Execute复杂、探索性的任务适合ReAct或Reflection。4. 实战使用 Claude Code 构建一个 Goal-based 数据查询智能体Claude Code这里指基于Claude API的编程框架或模式而非某个特定软件非常适合快速原型开发。下面我们构建一个智能体其目标是“从一份包含销售数据的混乱文本中找到第二季度的总销售额并判断是否比第一季度增长了10%以上。”4.1 环境准备与状态定义首先假设我们已经有了Claude的API访问权限。我们使用Python和Pydantic。# 安装必要库pip install pydantic anthropic import anthropic from pydantic import BaseModel from typing import List, Optional, Dict, Any import json import re # 初始化Claude客户端请替换你的API密钥 client anthropic.Anthropic(api_keyyour-api-key) # 定义状态模型 class SalesAnalysisState(BaseModel): goal: str raw_text: str # 原始混乱文本 cleaned_data: Optional[Dict[str, Any]] None # 清洗后的结构化数据 quarterly_sales: Optional[Dict[str, float]] None # 季度销售额如 {Q1: 1000, Q2: 1200} q2_total: Optional[float] None q1_total: Optional[float] None growth_rate: Optional[float] None judgment: Optional[str] None # 判断结果 history: List[str] [] # 记录每一步操作 error: Optional[str] None is_complete: bool False4.2 工具定义与实现我们为智能体设计几个专用工具。# 工具1提取并清洗数值数据 def extract_and_clean_numbers(text: str) - Dict[str, Any]: 从文本中提取所有类似金额的数字并尝试关联上下文如季度标识。 这是一个简化示例实际中可能需要更复杂的NLP。 state.history.append(f调用工具[extract_and_clean_numbers]输入文本长度{len(text)}) # 使用正则表达式查找数字和可能的前后关键词 pattern r(\bQ[1-4]\b|\b第一季度\b|\b第二季度\b|\bQ1\b|\bQ2\b)?[^0-9]*?(\d(?:,\d{3})*(?:\.\d{2})?)\s*(万元|元|美元|USD)? matches re.finditer(pattern, text, re.IGNORECASE) data_points [] for match in matches: period, value_str, unit match.groups() # 清洗数值字符串 value float(value_str.replace(,, )) # 简单单位换算示例 if unit and 万 in unit: value * 10000 data_points.append({period_hint: period, value: value, raw_match: match.group()}) # 非常简单的逻辑将数字按“季度”提示分组没有提示的单独存放 cleaned_data {Q1: [], Q2: [], Q3: [], Q4: [], unknown: []} for dp in data_points: period_key unknown if dp[period_hint]: if 1 in dp[period_hint] or 一 in dp[period_hint]: period_key Q1 elif 2 in dp[period_hint] or 二 in dp[period_hint]: period_key Q2 # ... 类似处理 Q3, Q4 cleaned_data[period_key].append(dp[value]) state.history.append(f工具返回找到数据点 {len(data_points)} 个按季度初步分组。) return cleaned_data # 工具2请求Claude进行智能解析 def ask_claude_for_interpretation(text: str, question: str) - str: 将文本和问题发给Claude请求其解析并回答。 state.history.append(f调用工具[ask_claude_for_interpretation]问题{question[:50]}...) try: message client.messages.create( modelclaude-3-sonnet-20240229, # 根据实际情况选择模型 max_tokens1000, messages[{ role: user, content: f请分析以下文本\n\n{text}\n\n问题{question}\n请直接给出答案不要解释过程。 }] ) answer message.content[0].text.strip() state.history.append(fClaude 回复{answer[:100]}...) return answer except Exception as e: state.history.append(f调用Claude失败{e}) return fError: {e} # 工具3计算增长率并判断 def calculate_growth_and_judge(q1_sales: float, q2_sales: float) - Dict[str, Any]: 计算季度环比增长率并做出判断。 if q1_sales 0: growth float(inf) else: growth (q2_sales - q1_sales) / q1_sales * 100 judgment 是 if growth 10 else 否 return {growth_rate: round(growth, 2), judgment: judgment}4.3 决策引擎与主循环实现我们采用一个混合决策引擎先用规则尝试如果不行则求助Claude。def decide_next_action(state: SalesAnalysisState) - str: 决策引擎根据当前状态决定下一步动作。 state.iteration_count 1 # 规则1检查是否已完成 if state.is_complete: return FINISH # 规则2检查错误 if state.error: return HANDLE_ERROR # 规则3检查迭代次数是否过多 if state.iteration_count 8: state.error 超过最大迭代次数 return HANDLE_ERROR # 基于状态的决策流 if state.cleaned_data is None: # 第一步清洗数据 return EXTRACT_DATA elif state.q2_total is None or state.q1_total is None: # 第二步如果清洗后数据仍无法确定季度总额则求助Claude if state.cleaned_data.get(Q2) and state.cleaned_data.get(Q1): # 如果能从清洗数据中直接求和 state.q2_total sum(state.cleaned_data[Q2]) state.q1_total sum(state.cleaned_data[Q1]) return CALCULATE_GROWTH else: # 数据不明确需要Claude介入解读 return ASK_CLAUDE_FOR_TOTALS elif state.growth_rate is None: # 第三步计算增长率 return CALCULATE_GROWTH else: # 所有步骤完成 state.is_complete True return FINISH def main_loop(initial_goal: str, raw_text: str): 主循环控制器。 # 初始化状态 state SalesAnalysisState(goalinitial_goal, raw_textraw_text, iteration_count0) print(f开始处理目标{state.goal}) while not state.is_complete and state.iteration_count 10: action decide_next_action(state) state.history.append(f迭代{state.iteration_count}决策动作{action}) if action EXTRACT_DATA: state.cleaned_data extract_and_clean_numbers(state.raw_text) elif action ASK_CLAUDE_FOR_TOTALS: question 请从上述文本中找出第一季度Q1的总销售额和第二季度Q2的总销售额只返回两个数字格式为Q1: [数字], Q2: [数字] answer ask_claude_for_interpretation(state.raw_text, question) # 解析Claude的答案 # 这里需要写一个简单的解析器来提取数字为节省篇幅假设解析成功并赋值 # 例如state.q1_total parsed_q1; state.q2_total parsed_q2 # 模拟解析成功 state.q1_total 1250000.0 # 假设值 state.q2_total 1450000.0 # 假设值 state.history.append(f从Claude解析得到 Q1: {state.q1_total}, Q2: {state.q2_total}) elif action CALCULATE_GROWTH: if state.q1_total and state.q2_total: result calculate_growth_and_judge(state.q1_total, state.q2_total) state.growth_rate result[growth_rate] state.judgment result[judgment] state.history.append(f计算完成增长率 {state.growth_rate}%判断 {state.judgment}) else: state.error 无法计算增长率缺少季度总额数据 elif action HANDLE_ERROR: print(f处理过程中遇到错误{state.error}) # 这里可以添加错误恢复逻辑比如重试、换用备用方案等 state.is_complete True # 或根据情况决定是否终止 elif action FINISH: state.is_complete True print(任务完成) break else: state.error f未知动作{action} action HANDLE_ERROR # 输出最终结果和历史 print(\n 最终状态 ) print(f目标{state.goal}) print(fQ1销售额{state.q1_total}) print(fQ2销售额{state.q2_total}) print(f增长率{state.growth_rate}%) print(f判断是否增长10%{state.judgment}) print(f是否完成{state.is_complete}) print(f总迭代次数{state.iteration_count}) print(\n 执行历史 ) for i, step in enumerate(state.history): print(f{i1}. {step}) # 模拟运行 if __name__ __main__: sample_text 第一季度销售报告显示一月销售额为120万元二月110万三月有所下滑至98万。 第二季度业绩回暖四月销售额达到130万元五月继续攀升至142万六月最终以155万元收官。 上半年总体表现良好。 main_loop(找出第二季度总销售额并判断是否比第一季度增长10%以上, sample_text)这个示例展示了一个完整的、可运行的Goal-based Loop智能体。它结合了规则引擎决策逻辑和LLM工具复杂解析有清晰的状态流转和历史记录。你可以看到循环的每一步都依赖于状态而决策函数是控制流程的核心。5. 高级技巧与避坑指南在实际开发中仅仅实现基础循环是远远不够的。下面分享一些能极大提升智能体可靠性和效率的高级技巧以及我踩过的一些坑。5.1 提示工程让LLM在循环中稳定输出LLM在循环中扮演“决策者”或“解析者”时其输出的不稳定性是最大挑战。你需要通过精心的提示词Prompt来约束它。为输出设计严格的格式要求LLM以特定格式如JSON、XML、纯文本的固定标记输出。这极大方便了后续的程序化解析。坏提示“告诉我第一季度销售额。”好提示“请从文本中提取第一季度总销售额。只输出一个数字不要任何其他文字。如果无法确定输出‘UNKNOWN’。示例输出1500000”提供清晰的示例Few-shot在提示词中给出一两个输入输出的例子能显著提升LLM遵循指令的能力。分而治之不要让一个LLM调用完成所有事情。将复杂任务分解成多个子任务每个子任务用一个简单的、格式固定的LLM调用来解决。例如先调用一次提取所有数字再调用一次对数字进行分类。设置明确的停止条件在Prompt中当LLM作为决策引擎时在Prompt里明确告诉它什么情况下应该停止。例如“如果你的分析表明已经获得了第二季度的明确总销售额且计算出了增长率请输出‘TASK_COMPLETE’。”实操心得对于关键决策点可以采用“自我验证”技巧。让LLM先输出一个答案再基于同一个问题但稍作修改的Prompt例如“请从另一个角度检查你刚才的答案是否正确”让它验证自己的输出。如果两次结果一致可信度就高很多。5.2 超时、重试与熔断机制网络请求、API调用总会失败。一个健壮的循环必须能处理这些异常。指数退避重试对于暂时性失败如网络超时、API限流不要立即放弃。实现一个重试逻辑并且每次重试的等待时间指数级增加如1秒、2秒、4秒、8秒。import time from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) def call_llm_with_retry(prompt): # 你的LLM调用代码 return client.messages.create(...)使用tenacity库可以优雅地实现重试循环超时为整个循环设置一个总超时时间。防止因逻辑错误或意外情况导致无限循环。import signal class TimeoutException(Exception): pass def timeout_handler(signum, frame): raise TimeoutException() signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(60) # 设置60秒超时 try: main_loop() except TimeoutException: print(循环执行超时) # 保存当前状态以便后续恢复或分析 finally: signal.alarm(0) # 取消闹钟熔断器模式如果某个工具如一个外部API连续失败多次暂时“熔断”对该工具的调用直接返回一个预设的降级结果或快速失败过一段时间再尝试恢复。这可以防止一个组件的故障拖垮整个系统。5.3 状态快照与可观测性调试一个运行中的智能体循环是痛苦的因为它的状态在不断变化。记录完整历史如前所述在状态中保存history列表记录每一步的决策、调用的工具、输入输出。这是事后分析问题的黄金记录。状态快照在每次循环迭代结束后将整个状态对象序列化如转换成JSON并保存到文件或日志系统。这样即使程序崩溃你也可以从最近的快照恢复运行。可视化工具可以考虑开发简单的可视化界面实时展示状态机的流转、工具调用链和历史记录。这对于演示和理解智能体行为非常有帮助。5.4 成本与性能优化LLM API调用是按Token收费的循环可能导致调用次数激增成本不可控。缓存对于相同的输入LLM的输出应该是确定的在温度0时。可以对LLM的请求和响应进行缓存。例如使用functools.lru_cache装饰器或外部缓存如Redis。from functools import lru_cache lru_cache(maxsize128) def cached_llm_call(prompt: str) - str: # 去重后的LLM调用 return call_llm(prompt)精简上下文每次调用LLM时只发送必要的上下文。避免将整个对话历史或庞大的状态全塞进去。可以设计一个函数来总结或筛选出与当前决策最相关的历史信息。设置预算上限在循环开始前估算单次任务的最大成本如最多调用LLM 10次每次平均消耗1000 Token。在循环中实时累计算消耗的Token数或调用次数接近上限时主动终止或转入降级处理流程。6. 常见问题排查与调试技巧即使设计得再完美实际运行中还是会遇到各种问题。下面是一个常见问题速查表以及我的调试心得。问题现象可能原因排查步骤与解决方案智能体陷入无限循环1. 停止条件定义模糊或永远无法满足。2. 决策逻辑有缺陷导致在两个状态间来回跳转。3. LLM决策引擎的Prompt没有明确要求停止。1.检查状态历史看循环在重复执行哪几步。如果总是在A-B-A说明状态在A和B之间震荡。2.强化停止条件除了目标达成增加最大迭代次数、超时时间等硬性限制。3.在Prompt中明确加入“如果你认为已经足够接近目标或者无法取得进展请输出‘STOP’。”LLM输出格式不符合预期导致解析失败1. Prompt指令不够清晰。2. LLM“自由发挥”添加了额外解释。1.使用结构化输出要求在Prompt中明确“请以JSON格式输出{“action”: “xxx”, “reason”: “xxx”}”。2.后处理清洗在解析LLM输出前用正则表达式或简单字符串查找提取关键部分增加容错性。3.采用“输出解析器”很多AI应用框架如LangChain提供了输出解析器Output Parser能强制将LLM输出匹配到预定格式。工具调用频繁失败1. 工具函数本身有bug或依赖服务不稳定。2. 传递给工具的参数格式错误。1.增加工具调用的异常捕获和日志记录失败时的输入参数。2.实现前验证在决策引擎调用工具前先简单验证参数是否在合理范围内如非空、类型正确。3.为工具设计降级方案例如网络搜索工具失败时可以转而查询本地知识库或返回一个提示信息。状态变得臃肿影响后续LLM调用速度每次循环都将完整历史记录作为上下文传给LLM导致Token数爆炸。1.状态摘要设计一个函数将冗长的历史记录总结成一段简洁的文本只保留关键决策和结果。2.滑动窗口只保留最近N条历史记录作为上下文。3.向量检索将历史记录存入向量数据库在需要时只检索与当前决策最相关的几条历史。智能体做出的决策明显“愚蠢”1. 提供给LLM决策的上下文信息不足或有误。2. Prompt没有引导LLM进行足够的“思考”。1.引入“思维链”在要求LLM做决策的Prompt中明确要求它分步推理。例如“请按以下步骤思考1. 分析当前目标... 2. 回顾已有信息... 3. 评估可用工具... 4. 做出决策。”2.丰富状态信息确保传递给决策引擎的状态包含了所有必要维度而不仅仅是原始数据。调试终极心法把智能体当成一个黑盒程序来调试。输入是初始状态和目标输出是最终结果和完整的历史记录。当结果不对时不要只盯着最后一步要从头到尾仔细阅读历史记录看智能体是在哪一步开始“跑偏”的。是状态信息缺失是工具返回了错误结果还是LLM基于不完整信息做出了错误推理通过历史记录你总能定位到问题发生的第一个异常点。构建一个稳定、高效的智能体循环是一个需要不断迭代和打磨的过程。从最简单的规则引擎开始逐步引入LLM的智能并为其套上“缰绳”清晰的规则、格式约束、停止条件你就能创造出真正能解决复杂问题的自动化助手。