1. 项目概述从“一锤子买卖”到“持续对话”的思维跃迁在AI Agent的开发实践中我们常常会陷入一个误区将每一次用户请求视为一次独立的、原子化的“交易”。用户输入一个指令Agent调用工具、生成回复然后流程结束一切归零。这种模式我称之为“一锤子买卖”式开发。它简单直接但问题在于它完全割裂了上下文Agent就像一个只有短期记忆的“金鱼”无法进行连贯的、有状态的复杂任务。比如用户说“帮我分析一下上个月的销售数据”Agent生成了图表用户接着说“对比一下这个月和上个月的趋势”在“一锤子买卖”模式下Agent已经“忘记”了上个月的数据是什么需要用户重新提供所有上下文体验极其割裂。这正是“主循环”概念要解决的核心痛点。所谓“主循环”并非指一段简单的while True代码而是指驱动Agent运行的、能够维持状态、管理记忆、协调工具调用并处理持续交互的核心控制逻辑。它是Agent的“骨架”和“中枢神经系统”决定了Agent如何感知、思考、行动并学习。一个设计精良的主循环能将一次孤立的用户查询转变为一个可持续执行、状态可继承的“回合”。在这个回合内Agent可以记住之前的对话、工具调用结果、用户偏好甚至从错误中学习调整策略从而实现真正意义上的智能协作。最近随着Claude Code等强大代码模型的普及以及开源社区对Agent框架如Hermes Agent的深入探索构建具备复杂主循环的Agent门槛正在降低。但工具易得思想难求。本文将深入拆解如何为你的Agent构建一个健壮、灵活的主循环让你摆脱“单次请求-响应”的桎梏打造出能进行深度、持续交互的智能体。无论你是想开发一个能陪你debug一整天的编程助手还是一个能分步骤帮你规划旅行、订票订酒店的私人秘书理解并实现主循环都是必经之路。2. 主循环的核心架构与设计哲学2.1 主循环的四大核心组件一个完整的Agent主循环远不止一个while循环包裹着对大语言模型的调用。它是一套精密的系统通常由以下几个核心组件协同工作状态管理器这是Agent的“记忆体”。它负责维护和管理整个交互回合的上下文状态。状态不仅包括原始的对话历史更重要的是结构化、向量化后的记忆以及当前任务的目标、已完成的步骤、中间结果、用户设定的约束条件等。一个高效的状态管理器需要解决记忆的存储、检索、更新和遗忘防止上下文过长问题。规划与决策引擎这是Agent的“大脑”。它基于当前状态和用户输入或环境反馈决定下一步要做什么。是直接回答用户问题还是需要调用某个工具如搜索、计算、写文件如果需要调用工具应该以什么参数调用这个引擎的核心通常是大语言模型本身通过精心设计的提示词Prompt或函数调用Function Calling机制引导模型进行任务分解和规划。工具执行器这是Agent的“手和脚”。它负责安全、可靠地执行决策引擎发出的工具调用指令。这包括加载工具定义、验证输入参数、在沙箱或安全环境中执行代码/命令、捕获执行结果和可能的异常。工具执行器的设计直接关系到Agent的安全性和可靠性。观察与学习模块这是Agent的“反馈系统”。它分析工具执行的结果成功、失败、部分成功评估当前行动是否朝着目标前进并可能据此调整策略或更新内部知识。例如如果调用搜索工具返回的结果不相关Agent可能会尝试重新构造查询词如果代码执行出错Agent会分析错误日志并尝试修复。2.2 设计哲学事件驱动与状态机在设计主循环时有两种主流范式轮询式和事件驱动式。早期的简单Agent多采用轮询式循环检查是否有新用户输入然后处理。但对于需要处理异步工具调用如一个耗时很长的数据分析任务或外部事件触发如定时任务、监控报警的复杂Agent事件驱动架构更为合适。事件驱动的主循环将整个交互过程视为一系列事件的流动。例如UserMessageEvent用户发送消息ToolCallEvent需要调用工具ToolResultEvent工具返回结果AgentResponseEventAgent生成回复ErrorEvent发生错误主循环的核心变成一个“事件总线”或“消息队列”不同组件监听自己关心的事件类型并触发相应的处理函数。这种设计解耦了各个组件使得系统更容易扩展和维护。例如你可以轻松地新增一个监听ToolResultEvent的组件专门用于将成功的结果记录到知识库中。更进一步我们可以用状态机的思维来建模Agent的整个生命周期。Agent可能处于IDLE等待输入、THINKING规划中、ACTING执行工具、OBSERVING处理结果、RESPONDING生成回复等状态。主循环负责驱动状态之间的转换并确保状态转换的逻辑是清晰和健壮的。例如从ACTING状态转换到OBSERVING状态时必须确保工具执行的结果已被妥善接收和处理。2.3 与常见框架的对比为何要“从轮子造起”市面上已有不少优秀的Agent框架如LangChain、LlamaIndex、AutoGen等它们都提供了不同层次的主循环抽象。使用这些框架可以快速上手。然而理解“主循环”的深层价值在于当框架提供的默认流程无法满足你的特定需求时你能知道如何定制和改造。例如LangChain的AgentExecutor提供了一个标准的“思考-行动-观察”循环。但对于需要复杂记忆检索比如基于当前对话内容从海量历史记录中精准找回相关片段的场景或者需要自定义工具执行优先级某些工具调用失败后应尝试备用方案而非直接报错的场景你就需要深入其循环内部进行调整。自己动手设计主循环的过程正是你深入理解Agent运作机理、掌握其“可塑性”的最佳途径。这就像学编程从使用高级框架到理解底层原理是一个必经的升华过程。3. 构建主循环的实操步骤与核心代码3.1 第一步定义清晰的状态数据结构一切始于状态。在写第一行循环代码之前我们必须先定义好Agent状态的数据结构。我强烈建议使用Pydantic这类数据验证库来定义它能确保状态的结构化和类型安全。from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional from datetime import datetime class AgentState(BaseModel): Agent的核心状态容器 # 会话元数据 session_id: str user_id: Optional[str] None created_at: datetime Field(default_factorydatetime.now) # 对话记忆原始 conversation_history: List[Dict[str, Any]] Field(default_factorylist) # 对话记忆向量化/摘要化用于长期记忆检索 memory_embeddings: Optional[List[float]] None memory_summary: Optional[str] None # 当前任务上下文 current_goal: Optional[str] None # 当前回合的顶层目标 sub_tasks: List[Dict[str, Any]] Field(default_factorylist) # 分解后的子任务列表 completed_tasks: List[Dict[str, Any]] Field(default_factorylist) # 已完成的任务 current_task_index: int 0 # 当前正在执行的子任务索引 # 工具调用上下文 available_tools: List[Dict[str, Any]] Field(default_factorylist) # 本回合可用的工具列表 last_tool_call: Optional[Dict[str, Any]] None # 上一次工具调用的记录 last_tool_result: Optional[Dict[str, Any]] None # 上一次工具调用的结果 # 环境与约束 constraints: List[str] Field(default_factorylist) # 用户给定的约束如“不要联网搜索” max_iterations: int 10 # 本回合最大循环次数防止死循环 iteration_count: int 0 # 当前已循环次数 # 自定义扩展字段 custom_data: Dict[str, Any] Field(default_factorydict)这个AgentState类定义了主循环将操作的核心数据。每次循环迭代我们都会读取和更新这个状态对象。3.2 第二步实现基础的事件驱动主循环骨架接下来我们实现一个基于简单事件驱动模型的主循环。这里我们使用一个字典来映射事件类型到处理函数。import asyncio from enum import Enum from typing import Callable, Awaitable class EventType(Enum): USER_MESSAGE user_message TOOL_CALL tool_call TOOL_RESULT tool_result AGENT_THINK agent_think AGENT_RESPOND agent_respond ERROR error class Event: def __init__(self, event_type: EventType, data: Any, state: AgentState): self.type event_type self.data data self.state state class AgentMainLoop: def __init__(self, llm_client, tools_registry): self.llm llm_client # 大语言模型客户端如OpenAI, Claude, DeepSeek等 self.tools tools_registry # 工具注册中心 self.state None self._event_handlers {event_type: [] for event_type in EventType} def register_handler(self, event_type: EventType, handler: Callable[[Event], Awaitable[None]]): 注册事件处理器 self._event_handlers[event_type].append(handler) async def _emit_event(self, event: Event): 触发事件调用所有注册的处理器 handlers self._event_handlers.get(event.type, []) for handler in handlers: await handler(event) async def run_for_session(self, initial_state: AgentState, user_input: str): 为一个会话启动主循环 self.state initial_state # 1. 初始化触发用户消息事件 await self._emit_event(Event(EventType.USER_MESSAGE, user_input, self.state)) # 2. 核心循环直到任务完成或达到迭代上限 while not self._is_session_complete() and self.state.iteration_count self.state.max_iterations: self.state.iteration_count 1 print(f[Loop Iteration {self.state.iteration_count}]) # 2.1 思考阶段触发思考事件让规划引擎工作 await self._emit_event(Event(EventType.AGENT_THINK, None, self.state)) # 2.2 检查是否需要行动调用工具 if self._needs_to_act(): # 触发工具调用事件 tool_call_spec self._plan_next_action() # 从状态中获取规划引擎决定的下一个动作 await self._emit_event(Event(EventType.TOOL_CALL, tool_call_spec, self.state)) # 模拟执行工具并获取结果 tool_result await self._execute_tool(tool_call_spec) # 触发工具结果事件 await self._emit_event(Event(EventType.TOOL_RESULT, tool_result, self.state)) # 2.3 生成回复如果当前子任务完成或需要与用户交互 if self._should_respond_now(): await self._emit_event(Event(EventType.AGENT_RESPOND, None, self.state)) break # 本次循环结束等待下次用户输入 # 3. 循环结束处理 if self.state.iteration_count self.state.max_iterations: print(警告达到最大迭代次数会话可能未完成。) return self.state def _is_session_complete(self): 判断当前会话回合是否完成 # 逻辑所有子任务完成且没有待解决的工具调用 return (self.state.current_task_index len(self.state.sub_tasks) and self.state.last_tool_call is None) def _needs_to_act(self): 判断当前是否需要调用工具 # 逻辑有未完成的子任务且上一个工具调用已处理完毕 return (self.state.current_task_index len(self.state.sub_tasks) and self.state.last_tool_result is not None) def _should_respond_now(self): 判断当前是否应该生成回复给用户 # 逻辑一个子任务完成或遇到需要用户确认的节点或发生错误 return (self.state.last_tool_result and self.state.last_tool_result.get(requires_user_input)) or self._is_session_complete() async def _execute_tool(self, tool_spec: dict): 模拟工具执行 # 实际项目中这里会调用真实的工具函数 tool_name tool_spec.get(name) print(f执行工具: {tool_name} with args: {tool_spec.get(arguments)}) # 模拟执行耗时 await asyncio.sleep(0.5) # 返回模拟结果 return {success: True, output: fTool {tool_name} executed successfully., requires_user_input: False} def _plan_next_action(self): 规划下一个动作简化版 # 在实际实现中这里会调用LLM进行任务规划和工具选择 # 此处返回一个模拟的工具调用规范 return {name: search_web, arguments: {query: latest AI news}}这个骨架展示了主循环如何围绕状态和事件运转。run_for_session方法是入口它初始化状态然后进入while循环。在循环中它依次触发THINK、ACT如果需要、OBSERVE、RESPOND如果需要等事件。具体的行为逻辑则封装在各个事件处理器中。3.3 第三步编写关键的事件处理器主循环的“血肉”在于事件处理器。我们以AGENT_THINK和TOOL_RESULT处理器为例。async def _setup_default_handlers(self): 设置默认的事件处理器 # 1. 思考处理器调用LLM进行规划和决策 async def think_handler(event: Event): state event.state # 构建给LLM的提示词包含历史、目标、可用工具等信息 prompt self._construct_planning_prompt(state) try: # 调用LLM这里以OpenAI格式为例实际可替换为Claude、DeepSeek等 response await self.llm.chat.completions.create( modelgpt-4, messages[{role: system, content: You are a planning assistant.}, {role: user, content: prompt}], toolsself._format_tools_for_llm(state.available_tools), # 将工具格式化为函数调用规范 tool_choiceauto ) message response.choices[0].message if message.tool_calls: # LLM决定调用工具 tool_call message.tool_calls[0] state.last_tool_call { id: tool_call.id, name: tool_call.function.name, arguments: json.loads(tool_call.function.arguments) } state.last_tool_result None # 清空上一次结果等待新结果 print(f规划决定调用工具 {tool_call.function.name}) else: # LLM决定直接回复或任务已完成 state.last_tool_call None state.agent_response_draft message.content print(f规划决定直接回复或任务结束) except Exception as e: # 触发错误事件 await self._emit_event(Event(EventType.ERROR, str(e), state)) self.register_handler(EventType.AGENT_THINK, think_handler) # 2. 工具结果处理器分析结果并更新状态 async def tool_result_handler(event: Event): state event.state tool_result event.data if tool_result.get(success): # 成功更新任务状态可能推进current_task_index state.completed_tasks.append({ task: state.sub_tasks[state.current_task_index], tool_call: state.last_tool_call, result: tool_result }) state.current_task_index 1 state.last_tool_result tool_result # 检查工具结果是否暗示需要新的规划例如结果出乎意料 if tool_result.get(output, ).lower().find(unexpected) ! -1: print(工具返回意外结果可能需要重新规划。) # 可以在这里设置一个标志让下一个循环重新思考 else: # 失败触发错误事件或尝试重试策略 await self._emit_event(Event(EventType.ERROR, fTool {state.last_tool_call[name]} failed: {tool_result.get(error)}, state)) self.register_handler(EventType.TOOL_RESULT, tool_result_handler) # 3. 响应处理器格式化最终回复给用户 async def respond_handler(event: Event): state event.state # 这里可以整合思考阶段的草稿、工具结果等生成友好的用户回复 final_response self._format_final_response(state) # 将回复添加到对话历史 state.conversation_history.append({role: assistant, content: final_response}) print(fAgent回复{final_response[:100]}...) self.register_handler(EventType.AGENT_RESPOND, respond_handler)通过注册这些处理器主循环的脉络就清晰了用户输入触发思考思考可能决定调用工具工具执行后结果被处理并可能更新任务状态最终在合适的时机生成回复。每个处理器只关心自己的职责符合单一职责原则。4. 高级模式复杂任务管理与记忆增强4.1 实现分层任务分解与回溯对于复杂目标如“开发一个简单的网页应用”单次规划往往不够。我们需要让Agent具备将大目标递归分解为子目标的能力并在执行中动态调整。这需要在状态中维护一个任务栈而不仅仅是任务列表。class HierarchicalAgentState(AgentState): 支持分层任务管理的状态 task_stack: List[Dict[str, Any]] Field(default_factorylist) # 任务栈栈顶是当前任务 def push_task(self, task: dict): 将新任务压入栈顶成为当前任务 self.task_stack.append(task) def pop_task(self) - Optional[dict]: 完成当前任务弹出栈顶返回上一级任务 if self.task_stack: completed self.task_stack.pop() self.completed_tasks.append(completed) return self.task_stack[-1] if self.task_stack else None return None property def current_task(self) - Optional[dict]: 获取当前正在执行的任务栈顶 return self.task_stack[-1] if self.task_stack else None在主循环的思考阶段LLM不仅可以选择工具还可以选择“分解任务”这个特殊的“元动作”。当选择分解任务时处理器会生成一系列子任务并将第一个子任务压入栈顶。当栈顶任务完成通过工具结果判断后自动弹出并继续执行栈中的下一个任务或者如果栈空了则回到顶层进行下一步规划。这种模式使得Agent能够处理非常复杂的、嵌套的项目。4.2 集成向量记忆与长期上下文管理随着对话轮数增加原始的对话历史会迅速耗尽模型的上下文窗口。解决方案是引入向量数据库实现长期记忆的存储和检索。记忆存储每当对话产生有价值的信息如用户提供的个人信息、解决的问题方案、学到的知识将其转换为文本片段生成向量嵌入存入向量数据库如Chroma、Pinecone、Qdrant并与当前会话ID关联。记忆检索在每次规划思考前将当前的用户查询和最近的对话上下文结合起来生成一个查询向量。用这个向量去向量数据库中搜索与本会话相关度最高的历史记忆片段。记忆注入将检索到的相关记忆片段作为系统提示词的一部分提供给LLM。这样LLM就能“回忆”起很久以前的对话内容实现真正的持续对话。# 简化的记忆检索处理器示例 async def memory_retrieval_handler(event: Event): if event.type ! EventType.AGENT_THINK: return state event.state # 构建检索查询当前目标 最近几句对话 query fGoal: {state.current_goal}. Recent context: {state.conversation_history[-3:]} # 从向量库检索相关记忆 relevant_memories await vector_db.similarity_search(query, filter{session_id: state.session_id}, k5) # 将记忆格式化后添加到state中供后续构造提示词使用 state.retrieved_memories [mem.page_content for mem in relevant_memories]通过这种方式Agent的“记忆”不再受限于短暂的上下文窗口而是可以扩展到成千上万条历史交互从而表现出更强的连贯性和个性化。5. 避坑指南与性能优化实战5.1 常见陷阱与解决方案循环失控无限循环现象Agent陷入“思考-调用同一个工具-得到相同结果-再思考”的死循环。根因规划逻辑有缺陷或者工具结果未能提供新的信息让Agent改变策略。解决方案硬性限制如我们代码中的max_iterations是必须的安全网。状态感知在状态中记录每个工具调用的历史。在规划时如果检测到最近三次行动完全一样则强制触发一个“重新评估”或“请求用户帮助”的机制。丰富工具结果确保工具失败时返回结构化的错误原因而不仅仅是“出错”。例如搜索工具没结果时可以返回“未找到相关信息建议尝试其他关键词”这能引导LLM改变策略。上下文污染与记忆混淆现象Agent混淆了不同会话或不同用户的信息。根因状态管理不隔离或者向量记忆检索时没有正确过滤会话ID。解决方案严格保证session_id和user_id的隔离。所有状态操作和记忆检索都必须附带这些ID作为过滤条件。为每个新对话创建一个全新的AgentState实例。工具调用安全风险现象Agent被诱导执行危险命令如rm -rf /。根因工具执行器没有做足够的输入验证和沙箱隔离。解决方案权限最小化每个工具只授予完成其功能所需的最小权限。输入验证与净化对LLM传来的参数进行严格的类型检查和内容过滤如禁止某些敏感关键词。沙箱执行对于执行代码、命令等高危操作必须在Docker容器或安全的子进程沙箱中运行并设置资源CPU、内存、时间限制。5.2 性能优化技巧异步化一切主循环中的I/O操作LLM调用、工具执行、数据库查询都应该使用异步async/await。这能极大提高Agent在并发处理多个会话时的吞吐量。我们的示例骨架已经采用了异步设计。流式响应与渐进式思考对于耗时较长的任务不要让用户干等。可以让Agent先输出“我正在思考...”或“我正在执行第一步...”然后以流式Streaming的方式逐步输出思考和行动过程。这不仅能提升用户体验也符合人类协作的直觉。缓存与索引工具结果缓存对于纯函数式、幂等的工具如计算器、单位换算对其输入参数进行哈希缓存结果。下次遇到相同参数直接返回缓存节省成本和时间。记忆索引优化向量数据库的索引设置如HNSW参数直接影响检索速度和精度。需要根据记忆条数和查询频率进行调优。对于超长文本的记忆可以先使用LLM进行摘要再存储摘要的向量能有效提升检索质量。降低LLM调用成本条件触发思考不是每次循环都必须调用昂贵的LLM进行完整规划。如果上一个工具结果非常明确地指示了下一步例如一个“成功/失败”的布尔值可以直接由规则引擎决定下一步动作跳过LLM调用。使用更小、更快的模型进行简单决策可以设计一个“路由模型”先用小模型如GPT-3.5 Turbo判断下一步是需要复杂规划还是简单回复再将复杂任务交给大模型如GPT-4。这被称为模型级联Model Cascading。构建一个健壮的主循环是打造高质量Agent应用的基础。它决定了Agent的智商上限通过规划能力和情商下限通过状态和记忆管理。从理解状态、事件、循环这些基本概念开始逐步融入分层任务、长期记忆等高级特性并时刻警惕安全与性能陷阱你就能搭建出真正强大、可持续对话的智能助手。这个过程没有银弹需要不断的迭代、测试和打磨但每一次对主循环的优化都会直接体现在Agent与用户交互体验的显著提升上。