从零构建自主循环AI Agent系统：核心组件、实战代码与工程化部署

1. 先搞清楚“会自己找活”的Agent Loop到底解决了什么

别再手动一条条给AI下指令了。如果你还在为每个任务单独写Prompt、等结果、再根据结果写下一个Prompt，那说明你的工作流还停留在“手动挡”阶段。一个能“自己找活”的Agent Loop系统，核心解决的就是任务自动化与决策闭环的问题。

它不是一个简单的脚本，也不是一个固定流程的自动化工具。它的价值在于，你只需要给定一个初始目标和边界规则，系统就能自动分解目标、规划步骤、执行任务、检查结果，并根据结果决定下一步是继续、调整还是终止。这个过程会循环进行，直到达成目标或触发终止条件。我团队用这套思路搭建的系统，已经稳定处理了长达三个月的日常运营、内容生成和数据分析任务，把我们从重复的Prompt工程中解放了出来。

所以，这篇文章适合两类人看：一是被大量重复性、有逻辑链条的AI任务缠身的运营或产品同学；二是希望将AI能力深度集成到业务流中的开发者。最关键的价值不是“自动化”，而是“目标驱动下的自主迭代”。下面，我就按实际搭建和踩坑的顺序，带你从零搭一个这样的系统。

2. 搭建前必须想清楚的四个核心组件

在动手写代码之前，必须先理清系统的骨架。一个能自主循环的Agent系统，通常离不开四个核心组件，缺一不可。很多人失败，就是因为一上来就埋头写“循环”，却忽略了组件之间的职责划分和数据流转。

2.1 任务规划与分解器（Planner）

这是系统的大脑。它的输入是你的终极目标（比如：“生成一份本季度市场分析报告”），输出是一个可执行的任务列表或流程图。它不关心具体怎么做，只关心“要做什么”以及“先做什么后做什么”。

• 关键能力：理解复杂目标、进行逻辑分解、处理任务间的依赖关系。
• 常见实现：可以用一个专门的LLM（大语言模型）来担任，Prompt里需要清晰定义输出格式（如JSON列表）。更复杂的场景可能需要图规划算法。
• 避坑点：不要让它分解出不可执行或定义模糊的子任务（如“分析数据”），必须分解为“获取XX平台近90天销售数据”这样的具体动作。

2.2 技能执行器（Executor）

这是系统的手和脚。它接收来自Planner的具体任务指令，调用对应的工具或API去完成。一个系统里可以有多个执行器，每个负责一类技能。

• 关键能力：精准调用工具、处理输入输出、捕获执行异常。
• 常见实现：封装好的函数、类方法，或专门用于工具调用的LLM（如利用ReAct、Function Calling框架）。
• 避坑点：执行器必须足够健壮，要有完善的错误处理和重试机制。网络超时、API限额、数据格式错误是常见故障点。

2.3 结果评估与状态检查器（Evaluator）

这是系统的眼睛和质检员。执行器干完活，干得怎么样？任务算成功了吗？是否需要重试？下一步该干嘛？这些判断由Evaluator完成。

• 关键能力：根据预定标准评估任务结果、判断任务状态（成功/失败/需调整）、为下一步决策提供依据。
• 常见实现：可以是规则引擎（如检查输出是否为空、是否包含关键词），也可以是另一个LLM（用于评估内容质量、逻辑一致性等）。
• 避坑点：评估标准必须明确、可量化。避免使用“感觉不错”这种模糊标准，而是“检查报告是否包含‘趋势’、‘建议’、‘数据来源’三个章节”。

2.4 工作流引擎与记忆体（Orchestrator & Memory）

这是系统的中枢神经和记忆。它负责串联以上三个组件，管理整个循环流程，并记住之前发生了什么。

• 工作流引擎：控制流程（规划->执行->评估->下一步决策），处理循环、分支和并发。
• 记忆体：存储任务历史、中间结果、上下文信息，防止Agent“失忆”，也是实现长期目标的关键。
• 常见实现：可以用LangGraph、AutoGen这类框架快速搭建工作流；记忆可以用向量数据库存储长期记忆，用简单变量或数据库存储当前会话状态。
• 避坑点：流程设计要避免死循环。记忆体要定期清理，防止上下文过长导致LLM性能下降或成本激增。

把这四个组件画在一张图上，明确它们之间的数据流（谁输出什么给谁），你的系统设计就完成了一半。

3. 从零开始：手把手搭建一个内容运营Loop

理论说再多不如动手。我们以一个真实的轻量级场景为例：自动化的社交媒体内容灵感生成与筛选系统。目标是：给定一个主题（如“AI编程助手”），系统能自动搜索近期热点、生成多条内容创意，并筛选出最优质的一条。

3.1 环境与工具准备

我们选择Python环境，利用现有框架降低开发复杂度。

# 基础环境，建议使用虚拟环境 python -m venv agent_loop_env source agent_loop_env/bin/activate # Linux/macOS # agent_loop_env\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain langgraph tavily-python

• openai/langchain: 用于调用LLM（如GPT-4）作为我们Planner和Evaluator的“大脑”。
• langgraph: LangChain官方的工作流编排框架，非常适合构建有状态、可循环的Agent系统。
• tavily-python: 一个搜索API工具，作为Executor的技能之一。你也可以换成SerpAPI或其他。

注意：你需要准备好对应API的密钥，并设置环境变量。

export OPENAI_API_KEY="your_key" export TAVILY_API_KEY="your_key"

3.2 第一步：定义状态与构建技能（Executor）

在LangGraph中，我们首先定义整个工作流需要共享的“状态”。

from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): """定义工作流状态，所有节点都读写这个状态字典。""" topic: str # 输入主题 search_results: List[str] # 搜索到的信息 content_ideas: List[str] # 生成的内容创意 evaluated_ideas: List[dict] # 评估后的创意（带分数） final_idea: str # 最终选出的最佳创意

接下来，构建我们的第一个技能：网络搜索执行器。

from langchain_community.tools.tavily_search import TavilySearchResults # 初始化搜索工具 search_tool = TavilySearchResults(max_results=3) # 限制结果数量，控制成本 def search_node(state: AgentState): """执行搜索，将结果存入状态。""" print(f“正在搜索主题：{state[‘topic’]}”) try: results = search_tool.invoke({“query”: f”{state[‘topic’]} latest trends news 2024”}) # 提取摘要信息 search_info = [f”{r[‘title’]}: {r[‘content’]}” for r in results] return {“search_results”: search_info} except Exception as e: print(f“搜索失败：{e}”) return {“search_results”: [“搜索暂时不可用”]}

这个函数就是一个简单的Executor。它接收状态中的topic，调用搜索工具，将结果格式化后存回状态。

3.3 第二步：构建规划与创意生成节点（Planner + Executor）

这里我们将规划和执行合并在一个节点中，让LLM根据搜索结果为给定主题生成内容创意。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI(model=“gpt-4-turbo-preview”) # 使用能力较强的模型进行创意生成 def generate_ideas_node(state: AgentState): """基于搜索结果为主题生成多条内容创意。""" search_context = “\n”.join(state[‘search_results’]) prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个资深社交媒体内容策划。根据提供的搜索信息，为给定主题构思吸引人的内容创意。”), (“human”, “”” 主题：{topic} 相关搜索信息： {search_context} 请为该主题生成5个不同的内容创意（如推文思路、短视频脚本开头、博客文章角度等）。 每个创意用一句话清晰描述，直接以‘- ’开头列出。 “””) ]) chain = prompt | llm response = chain.invoke({“topic”: state[‘topic’], “search_context”: search_context}) # 解析LLM返回的文本，提取创意列表 ideas_text = response.content ideas_list = [line.strip(“- “).strip() for line in ideas_text.split(‘\n’) if line.startswith(‘-’)] # 只取前5个，确保数量 ideas_list = ideas_list[:5] print(f“已生成创意：{ideas_list}”) return {“content_ideas”: ideas_list}

这个节点充当了“规划+执行”的角色：它“规划”出5个创意方向，并“执行”了生成创意的动作。

3.4 第三步：构建评估与筛选节点（Evaluator）

生成了一堆创意，哪个最好？我们需要另一个LLM来担任评委。

def evaluate_and_select_node(state: AgentState): """评估所有内容创意，并选出最佳的一个。""" if not state[‘content_ideas’]: return {“final_idea”: “未生成有效创意”, “evaluated_ideas”: []} ideas_text = “\n”.join([f”{i+1}. {idea}” for i, idea in enumerate(state[‘content_ideas’])]) prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个挑剔的社交媒体内容主编。你的任务是根据‘新颖性’、‘传播潜力’、‘与主题相关性’三个维度，为每个创意打分（1-10分），并选出总分最高的一个。”), (“human”, “”” 主题：{topic} 待评估的创意列表： {ideas_text} 请按以下格式输出你的评估结果： 1. 首先，为每个创意输出一行：`创意X | 新颖性: A | 传播力: B | 相关性: C | 总分: D` 2. 最后一行输出：`最佳创意：X` (X为创意编号) 请确保输出严格遵循此格式。 “””) ]) chain = prompt | llm response = chain.invoke({“topic”: state[‘topic’], “ideas_text”: ideas_text}) # 解析评估结果（这是一个简化的解析，实际应用需要更健壮的解析逻辑） lines = response.content.split(‘\n’) evaluated = [] best_idea_num = None best_idea_text = “” for line in lines: if ‘|’ in line: evaluated.append(line.strip()) if line.startswith(‘最佳创意：’): try: best_idea_num = int(line.replace(‘最佳创意：’, ‘’).strip()) except: pass # 根据编号找到最佳创意文本 if best_idea_num and 1 <= best_idea_num <= len(state[‘content_ideas’]): best_idea_text = state[‘content_ideas’][best_idea_num - 1] print(f“评估完成。最佳创意是：{best_idea_text}”) return {“evaluated_ideas”: evaluated, “final_idea”: best_idea_text}

这个Evaluator节点引入了决策逻辑。系统不再只是机械执行，而是能基于一套标准做出选择。

3.5 第四步：用LangGraph组装循环工作流

现在，我们把三个节点组装起来，并决定它们的执行顺序。目前这是一个简单的线性流程：搜索 -> 生成 -> 评估。但Graph的强大之处在于可以轻松添加循环。

from langgraph.graph import StateGraph, END # 创建工作流构建器 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node(“search”, search_node) workflow.add_node(“generate_ideas”, generate_ideas_node) workflow.add_node(“evaluate”, evaluate_and_select_node) # 设置边的连接关系（定义执行顺序） workflow.set_entry_point(“search”) workflow.add_edge(“search”, “generate_ideas”) workflow.add_edge(“generate_ideas”, “evaluate”) workflow.add_edge(“evaluate”, END) # 编译图 app = workflow.compile()

至此，一个最简单的单向Agent工作流就完成了。你可以运行它：

# 定义初始状态 initial_state = AgentState(topic=“AI编程助手”, search_results=[], content_ideas=[], evaluated_ideas=[], final_idea=“”) # 执行图 final_state = app.invoke(initial_state) print(f“\n最终结果：{final_state[‘final_idea’]}”)

4. 从“流水线”升级到“真循环”：让系统自己判断何时停止

上面的例子只是一个固定流程的“流水线”。如何让它变成“会自己找活”的Loop？关键在于在评估节点后增加一个判断逻辑，决定是结束循环，还是开启新一轮任务。

假设我们的新目标是：生成一个“足够好”的创意，标准是评估总分超过24分（满分30）。如果达不到，就调整方向重新生成。

4.1 修改状态与评估节点

首先，状态里需要记录历史尝试和当前分数。

class LoopAgentState(TypedDict): topic: str search_results: List[str] current_idea: str # 当前生成的单个创意 current_score: int # 当前创意的评估总分 attempt_count: int # 尝试次数 best_idea_so_far: str # 目前最好的创意 best_score_so_far: int is_satisfied: bool # 是否满足条件，用于控制循环

然后，修改评估节点，使其能解析出具体分数，并判断是否达标。

def evaluate_with_condition_node(state: LoopAgentState): """评估当前创意，并判断是否满足循环终止条件。""" # ... (前面的评估逻辑类似，但改为评估单个state[‘current_idea’]) # 假设我们从LLM评估结果中解析出了分数 `total_score` total_score = 22 # 示例：假设本次评估得分为22 state[‘current_score’] = total_score state[‘attempt_count’] += 1 # 更新历史最佳 if total_score > state[‘best_score_so_far’]: state[‘best_score_so_far’] = total_score state[‘best_idea_so_far’] = state[‘current_idea’] # 循环终止条件判断 if total_score >= 24: print(f“创意达标（分数：{total_score}），循环终止。”) state[‘is_satisfied’] = True elif state[‘attempt_count’] >= 5: # 设置最大尝试次数，防止无限循环 print(f“已达到最大尝试次数（{state[‘attempt_count’]}），循环终止。”) state[‘is_satisfied’] = True else: print(f“创意未达标（分数：{total_score}），将进行第{state[‘attempt_count’] + 1}次尝试。”) state[‘is_satisfied’] = False return state

4.2 设计循环逻辑与条件边

在LangGraph中，我们使用条件边来实现循环。

from langgraph.graph import StateGraph, END loop_workflow = StateGraph(LoopAgentState) loop_workflow.add_node(“search”, search_node) # 复用搜索节点，或修改为每次生成新查询 loop_workflow.add_node(“generate_one_idea”, generate_one_idea_node) # 新节点：每次生成一个创意 loop_workflow.add_node(“evaluate_condition”, evaluate_with_condition_node) loop_workflow.set_entry_point(“search”) # 定义边 loop_workflow.add_edge(“search”, “generate_one_idea”) loop_workflow.add_edge(“generate_one_idea”, “evaluate_condition”) # 关键：从评估节点出来的条件边 def should_continue(state: LoopAgentState): """根据评估结果，决定下一步是继续循环还是结束。""" if state[‘is_satisfied’]: return “end” # 满足条件，结束 else: return “generate_one_idea” # 不满足条件，返回去重新生成创意 # 注意：这里也可以选择返回“search”重新搜索，实现更复杂的循环逻辑 loop_workflow.add_conditional_edges( “evaluate_condition”, should_continue, # 条件判断函数 { “end”: END, “generate_one_idea”: “generate_one_idea” } ) # 从‘generate_one_idea’到‘evaluate_condition’的边已经在前面添加了，这样就形成了一个环。 loop_app = loop_workflow.compile()

现在，这个系统就具备了“自主循环”的能力：生成 -> 评估 -> 不达标 -> 再生成 -> 再评估……直到达标或超过尝试次数。这就是“会自己找活”的雏形。

5. 投入生产前必须处理的五个关键问题

把Demo跑通只是第一步。要让这样的Loop系统稳定运行三个月，你必须解决以下五个工程化问题。

5.1 错误处理与鲁棒性

Agent系统涉及大量外部调用（LLM API、搜索API、数据库），网络波动、服务限流、响应格式异常随时可能发生。

• 策略：在每个可能失败的节点（尤其是Executor）加入重试机制（如tenacity库）和降级方案。
• 示例：搜索失败时，是返回缓存数据、使用备用搜索引擎，还是将任务标记为“需人工干预”并记录到日志？必须在设计时就定义好。
• 建议：使用try...except捕获具体异常，并根据异常类型决定重试、跳过还是告警。不要用一个except Exception吞掉所有错误。

5.2 状态管理与持久化

内存中的状态在程序重启后会丢失。对于需要长时间运行或处理重要任务的Loop，必须将状态持久化。

• 策略：使用数据库（如SQLite、PostgreSQL）或文件系统来保存工作流状态。LangGraph本身支持将检查点（Checkpoint）持久化。
• 操作：在每次状态变更后，将关键的AgentState序列化（如转成JSON）并存储。系统重启时，可以从最后一个成功的检查点恢复执行。
• 避坑：注意存储敏感信息（如API返回的原始数据）可能带来的安全和成本问题，必要时只存储摘要或索引。

5.3 成本与性能监控

自主循环可能在你不知情的情况下消耗大量API调用。一个失控的循环可能导致巨额账单。

• 策略：为每个循环设置明确的预算和超时。
  • 预算：限制最大尝试次数、总Token消耗或总API调用次数。
  • 超时：为整个工作流或单个节点设置执行超时。
• 监控：在关键节点埋点，记录每次LLM调用的Token数、耗时、费用估算。可以使用LangSmith等LLM应用监控平台。
• 建议：在开发环境使用较便宜的模型（如GPT-3.5-turbo），上线前再切换。对于评估类任务，可以尝试使用小模型或规则引擎来降低成本。

5.4 任务粒度的控制与人工介入

全自动不等于完全不需要人。系统应该支持“人在环路”。

• 策略：在关键决策点（如评估结果处于临界值、循环次数过多、成本超预算）设置“中断点”，将状态和上下文发送给人工审核（如通过邮件、Slack消息），等待批准后再继续。
• 实现：可以在工作流中插入一个“human_review”节点，该节点暂停工作流，等待外部输入（如一个管理后台的审批操作）后再决定下一步走向。
• 价值：这不仅能防止错误扩散，也是收集反馈、优化系统的重要途径。

5.5 可观测性与调试

当系统行为不符合预期时，你需要快速知道“卡在哪了”“为什么这么决策”。

• 必须记录：
  1. 完整的执行轨迹：每个节点的输入/输出。
  2. LLM的原始请求与响应：这是理解Agent“思维过程”的关键。
  3. 工具调用详情：调用了什么API，传了什么参数，返回了什么。
  4. 循环控制日志：每次评估的分数、是否满足条件、下一步方向。
• 工具：同样推荐使用LangSmith，它能为LangGraph应用提供可视化的执行轨迹图，极大提升调试效率。自建的话，需要设计结构化的日志系统。

6. 从“玩具”到“生产”：三个月的实战经验提炼

运行三个月后，我们总结出几条超越具体代码的通用经验。

第一，目标定义比算法选择更重要。在搭建Loop之初，必须花80%的时间来厘清：你的“目标”是否可以被清晰评估？你给的“边界规则”是否无歧义？一个模糊的目标（如“提升品牌影响力”）会导致评估器失效，循环要么早早终止，要么无限空转。务必把目标拆解成系统可以量化判断的指标。

第二，让循环“慢下来”往往比“跑得快”更重要。初期我们追求全速自动化，但后来发现，在关键节点（如生成重要内容、做出分类决策）后强制加入一个短暂的“冷却期”或“二次确认”逻辑，能有效避免因LLM的随机性导致的错误累积。这类似于给高速运转的机器加上离合器。

第三，系统的“健康度”需要持续喂养。Agent Loop不是一次搭建终身受用的。业务在变，网络信息在变，LLM本身也在变。你需要定期：

• 检查评估标准：当初定的打分标准还适用吗？
• 审核失败案例：系统在哪些任务上总是失败？是工具问题、Prompt问题还是流程问题？
• 更新知识库/搜索源：Executor所依赖的外部信息源是否依然可靠、全面？

最后，也是最重要的心态转变：从“操作员”变为“教练”。搭建这类系统后，你的核心工作不再是亲自处理每一个任务，而是设计更好的目标、提供更有效的工具（技能）、制定更合理的规则（评估标准），并持续训练和调整你的“AI团队”。当系统能稳定自主地处理80%的常规工作时，你才有精力去攻克那20%更复杂、更有价值的新问题。这才是“会自己找活”的Agent Loop带来的真正解放。