从零构建自主循环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-pythonopenai/langchain: 用于调用LLM如GPT-4作为我们Planner和Evaluator的“大脑”。langgraph: LangChain官方的工作流编排框架非常适合构建有状态、可循环的Agent系统。tavily-python: 一个搜索API工具作为Executor的技能之一。你也可以换成SerpAPI或其他。注意你需要准备好对应API的密钥并设置环境变量。export OPENAI_API_KEYyour_key export TAVILY_API_KEYyour_key3.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_results3) # 限制结果数量控制成本 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”{i1}. {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 state4.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 可观测性与调试当系统行为不符合预期时你需要快速知道“卡在哪了”“为什么这么决策”。必须记录完整的执行轨迹每个节点的输入/输出。LLM的原始请求与响应这是理解Agent“思维过程”的关键。工具调用详情调用了什么API传了什么参数返回了什么。循环控制日志每次评估的分数、是否满足条件、下一步方向。工具同样推荐使用LangSmith它能为LangGraph应用提供可视化的执行轨迹图极大提升调试效率。自建的话需要设计结构化的日志系统。6. 从“玩具”到“生产”三个月的实战经验提炼运行三个月后我们总结出几条超越具体代码的通用经验。第一目标定义比算法选择更重要。在搭建Loop之初必须花80%的时间来厘清你的“目标”是否可以被清晰评估你给的“边界规则”是否无歧义一个模糊的目标如“提升品牌影响力”会导致评估器失效循环要么早早终止要么无限空转。务必把目标拆解成系统可以量化判断的指标。第二让循环“慢下来”往往比“跑得快”更重要。初期我们追求全速自动化但后来发现在关键节点如生成重要内容、做出分类决策后强制加入一个短暂的“冷却期”或“二次确认”逻辑能有效避免因LLM的随机性导致的错误累积。这类似于给高速运转的机器加上离合器。第三系统的“健康度”需要持续喂养。Agent Loop不是一次搭建终身受用的。业务在变网络信息在变LLM本身也在变。你需要定期检查评估标准当初定的打分标准还适用吗审核失败案例系统在哪些任务上总是失败是工具问题、Prompt问题还是流程问题更新知识库/搜索源Executor所依赖的外部信息源是否依然可靠、全面最后也是最重要的心态转变从“操作员”变为“教练”。搭建这类系统后你的核心工作不再是亲自处理每一个任务而是设计更好的目标、提供更有效的工具技能、制定更合理的规则评估标准并持续训练和调整你的“AI团队”。当系统能稳定自主地处理80%的常规工作时你才有精力去攻克那20%更复杂、更有价值的新问题。这才是“会自己找活”的Agent Loop带来的真正解放。