1. 先搞清楚“Agent停了”到底是谁的问题

当你满怀期待地启动一个AI Agent，让它去处理一个需要多步骤、长时间运行的任务，比如整理一份周报、分析一批数据或者监控一个系统，结果发现它刚开了个头，或者执行到一半，就突然“停”了，没有任何错误提示，也没有继续执行。这种“任务未完成，Agent已下班”的情况，是很多人在初次接触Agent开发或使用时会遇到的典型困惑。

这个问题，尤其是在使用像Pi这类强调Goal（目标）导向的Agent框架时，会显得格外突出。因为你的直觉是：我给了它一个明确的目标（Goal），它就应该像人一样，不达目的不罢休，持续工作直到完成。但现实是，Agent的“大脑”——也就是驱动它的大语言模型（LLM）——本质上是一个“单步思考者”。它每次被调用，只产生一次输出（一个动作、一段文本）。如果没有一个外部的“监督员”或“循环机制”来反复调用它，并检查目标完成状态，它自然就“停”了。

所以，核心矛盾在于：我们期望的“持续执行”是业务逻辑，而Agent框架提供的“单步推理”是基础能力。框架的任务，就是通过一套机制（比如Pi的/goal），来填补这个鸿沟。如果你没理解这套机制，就会觉得Agent“不听话”或“有bug”。这篇文章，我们就以Pi框架的Goal机制为切入点，彻底拆解Agent持续执行的原理、配置和避坑点。无论你是用Pi、Hermes、AutoGPT还是其他Agent框架，这套思路都是通用的。

2. Pi的/goal：不只是个启动命令，而是任务生命周期管理器

很多人把/goal简单地理解为一个“开始任务”的指令，这其实只对了一半。在Pi框架的设计哲学里，/goal是一个任务生命周期的入口和协调者。它至少承担了三个关键角色：

1. 目标解析与规划器：接收你设定的自然语言目标（如“分析上周服务器日志，找出错误趋势”），并尝试将其分解为一系列可执行的子步骤（Steps）。
2. 执行引擎的触发器：根据规划，按顺序或条件触发具体的技能（Skills）或动作（Actions）去执行每个子步骤。
3. 完成状态的判断者：在每个子步骤执行后，评估当前状态是否已经达成了最初设定的目标。如果没有，就继续循环；如果达成，则优雅终止。

这个机制，就是持续执行的核心。Agent之所以会“停”，往往是因为这个循环链条在某个环节断掉了。下面我们通过一个最简单的例子，来看这个链条是如何工作的。

2.1 一个最小化的持续执行流程

假设我们有一个非常简单的Pi Agent，它只有一个技能：search_web（搜索网页）。我们的目标是：“搜索并总结今天关于AI Agent的最新新闻”。

理想中的持续执行流程应该是这样的：

1. 你： 向Agent发送指令/goal 搜索并总结今天关于AI Agent的最新新闻。
2. Pi的/goal处理器：
   • 解析： 理解这是一个“搜索”+“总结”的复合目标。
   • 规划： 生成第一步计划：“调用search_web技能，关键词为‘AI Agent 最新新闻 今天’”。
3. 执行引擎： 调用search_web技能，获得搜索结果（一段文本）。
4. /goal处理器（再次介入）：
   • 判断： 检查当前结果。“只有原始搜索结果，没有总结。目标未完成。”
   • 规划： 生成第二步计划：“基于上一步的搜索结果，调用总结能力（可能是另一个技能，或LLM本身的文本摘要能力），生成一份简洁的总结报告。”
5. 执行引擎： 调用总结功能，生成总结报告。
6. /goal处理器（最后检查）：
   • 判断： 检查当前结果。“已生成总结报告。目标‘搜索并总结’已完成。”
   • 终止： 停止循环，返回最终报告给你。

在这个过程中，/goal机制像一位项目经理，在每一步执行后都问自己：“我们最初要的东西，现在齐了吗？” 如果没齐，就安排下一个任务；如果齐了，就宣布项目结束。

2.2 为什么链条会断？常见的“停车点”

理解了流程，就能定位“停车”的原因。链条断裂通常发生在以下几个点：

• 规划阶段就失败了：LLM无法将你的模糊目标分解成明确的步骤。比如，你让它“让公司变得更好”，它可能直接回复“这是一个伟大的目标，但我需要更具体的指令”，然后就没有然后了。解决方案：给出明确、可操作、单一步骤能处理的目标。从“写周报”细化到“从Jira获取我本周已关闭的任务，生成Markdown格式的列表”。
• 执行步骤时出错：某个技能执行失败，抛出了异常，而Agent没有错误处理或重试机制。比如，search_web技能依赖的API密钥失效了。解决方案：为技能添加健壮的错误处理（try-catch），并在/goal的配置中考虑失败重试或备用方案。
• 状态判断逻辑有误：这是最隐蔽也最常见的问题。/goal用来判断“目标是否完成”（goal_complete）的逻辑写错了。比如，上面的例子中，如果判断逻辑是“只要调用了search_web就算完成”，那么Agent在第一步之后就会停止，不会进入总结步骤。解决方案：仔细审查和测试你的goal_complete条件判断函数。
• 资源或权限限制：任务运行时间过长、内存消耗过大被系统中断，或者访问某些资源没有权限。解决方案：监控Agent运行时的资源使用情况，确保它有必要的权限和合理的运行超时设置。
• 框架配置或版本问题：你可能没有正确启用或配置持续执行所需的组件（如工作流引擎、记忆模块）。或者框架本身有Bug。解决方案：查阅官方文档，确认持续执行是否为默认开启，或需要特定配置。关注社区和Issue列表。

3. 动手配置：让Pi Agent真正“跑起来”

理论说再多，不如动手调一遍。下面我们以一个更具体的场景为例，展示如何在Pi框架中配置一个具备基本持续执行能力的Agent。我们的目标是：让Agent读取指定目录下的所有文本文件，并统计每个文件的行数，最后输出一份报告。

3.1 环境与依赖准备

首先，确保你的环境已经就绪。Pi通常是一个Python框架。

# 假设你已经有了Python环境 pip install pi-framework # 这里用`pi-framework`作为示例包名，请替换为实际名称 # 可能还需要安装一些额外的依赖，比如文件处理库 pip install pathlib

关键点： 安装后，第一件事不是直接写Agent，而是跑通官方提供的一个最小示例（Hello World），确认基础环境（Python版本、依赖包、网络）没问题。很多“启动即失败”的问题都出在这一步。

3.2 定义核心技能（Skill）

Agent要持续工作，离不开技能。我们先定义一个最简单的文件行数统计技能。

# file_line_counter.py import os class FileLineCounterSkill: name = "file_line_counter" description = "统计指定文件的行数" def execute(self, file_path: str) -> dict: """执行技能：读取文件并统计行数""" try: if not os.path.exists(file_path): return {"success": False, "error": f"文件不存在: {file_path}", "line_count": 0} with open(file_path, 'r', encoding='utf-8') as f: lines = f.readlines() line_count = len(lines) return {"success": True, "file_path": file_path, "line_count": line_count} except Exception as e: return {"success": False, "error": str(e), "file_path": file_path, "line_count": 0}

为什么这么写：技能返回结构化的结果（包含成功标志、数据、错误信息），这便于后续的/goal处理器进行判断和决策。统一的返回格式是Agent技能设计的良好实践。

3.3 构建Goal与持续执行逻辑

这是最关键的一步。我们需要创建一个Goal，它知道要做什么（调用技能），以及如何判断自己做完了。

# my_agent.py import os from pi_framework import Agent, Goal # 再次提醒，类名根据实际框架调整 from file_line_counter import FileLineCounterSkill class CountLinesInDirectoryGoal(Goal): name = "count_lines_in_directory" description = "统计一个目录下所有文本文件的行数" def __init__(self, target_directory: str): super().__init__() self.target_directory = target_directory self.results = [] # 用来存储每个文件的结果 self.files_to_process = [] # 待处理文件列表 self.current_file_index = 0 # 当前处理到哪个文件 def on_goal_start(self): """Goal开始时的初始化：列出所有文件""" print(f"[Goal] 开始处理目录: {self.target_directory}") try: all_files = os.listdir(self.target_directory) # 过滤出.txt文件作为示例 self.files_to_process = [ os.path.join(self.target_directory, f) for f in all_files if f.endswith('.txt') ] print(f"[Goal] 找到 {len(self.files_to_process)} 个待处理的.txt文件") if not self.files_to_process: print("[Goal] 没有找到.txt文件，目标提前完成。") self.mark_completed() # 一个重要方法：标记目标完成 except Exception as e: print(f"[Goal] 初始化失败: {e}") self.mark_failed() # 标记目标失败 def get_next_step(self): """获取下一个要执行的步骤。这是持续执行的核心驱动方法。""" if self.current_file_index >= len(self.files_to_process): # 所有文件都处理完了，没有下一个步骤 return None next_file = self.files_to_process[self.current_file_index] self.current_file_index += 1 # 返回一个步骤描述，告诉Agent执行什么技能，传入什么参数 return { "skill_name": "file_line_counter", "inputs": {"file_path": next_file}, "description": f"统计文件行数: {next_file}" } def on_step_result(self, step_description: dict, result: dict): """处理每个步骤执行后的结果""" file_path = step_description["inputs"]["file_path"] if result.get("success"): line_count = result["line_count"] self.results.append({"file": file_path, "lines": line_count}) print(f"[Step] 成功: {file_path} -> {line_count} 行") else: error = result.get("error", "未知错误") self.results.append({"file": file_path, "error": error}) print(f"[Step] 失败: {file_path} -> {error}") def is_goal_complete(self) -> bool: """判断Goal是否完成。这是决定Agent是否‘停车’的关键函数。""" # 完成条件：1. 所有文件都已尝试处理（无论成功失败）；2. 或者初始化时就发现没有文件。 all_processed = (self.current_file_index >= len(self.files_to_process)) return all_processed or self.completed # self.completed 可能在 on_goal_start 中被标记 def on_goal_end(self): """Goal结束时的收尾工作：生成报告""" print("\n" + "="*50) print("[Goal] 任务完成！生成报告：") total_files = len(self.results) success_files = len([r for r in self.results if "lines" in r]) total_lines = sum([r["lines"] for r in self.results if "lines" in r]) print(f" 处理文件总数: {total_files}") print(f" 成功统计文件数: {success_files}") print(f" 失败文件数: {total_files - success_files}") print(f" 总行数: {total_lines}") print("="*50) # 这里可以将self.results保存到文件或数据库

逐段解析：

• on_goal_start: 这是任务起点。在这里做准备工作（如列出文件）。如果一开始条件就不满足（如目录为空），必须立即调用mark_completed()或mark_failed()，否则Agent会等待一个不存在的“下一步”，看起来就像卡住了。
• get_next_step:这是引擎的心脏。只要这个方法返回一个有效的步骤描述，Agent就会继续执行。当返回None时，引擎会去检查is_goal_complete。我们的逻辑是：按顺序返回每个文件，直到所有文件都返回过。
• on_step_result: 处理每个技能执行后的结果。在这里收集数据、记录日志。即使某个步骤失败，也不要让整个Goal崩溃，而是记录错误，继续下一个。这体现了持续执行的鲁棒性。
• is_goal_complete:最重要的判断函数。我们的逻辑是“所有文件都已尝试处理”。框架会反复调用这个函数。只有当它返回True时，Agent才会真正“停车”并进入on_goal_end。如果逻辑写错（比如写成“第一个文件成功就完成”），Agent就会过早停止。
• on_goal_end: 最终报告。在这里汇总结果、清理资源。

3.4 组装并运行Agent

# main.py from my_agent import CountLinesInDirectoryGoal from file_line_counter import FileLineCounterSkill # 假设Pi框架的Agent核心类叫`PiAgent` from pi_framework import PiAgent def main(): # 1. 创建Agent实例 agent = PiAgent(name="文件分析小助手") # 2. 注册技能 agent.register_skill(FileLineCounterSkill()) # 3. 指定要处理的目录 target_dir = "./data" # 假设你有一个名为data的目录，里面放了些.txt文件 # 4. 创建并设置Goal goal = CountLinesInDirectoryGoal(target_directory=target_dir) agent.set_current_goal(goal) # 5. 启动Agent执行循环 print("启动Agent执行Goal...") try: agent.run() # 框架的run方法会内部循环调用 get_next_step, 执行技能，检查 is_goal_complete print("Agent运行结束。") except KeyboardInterrupt: print("\n用户中断执行。") except Exception as e: print(f"Agent运行过程中发生未捕获异常: {e}") if __name__ == "__main__": main()

运行这个脚本，你应该能看到Agent逐个处理文件，直到所有文件处理完毕，然后打印汇总报告。这就是一个完整的、不会中途“停车”的持续执行过程。

4. 深度排查：当Agent依然“停车”时，你的检查清单

即使按照上面的模板写了，Agent可能还是会出问题。别急着怀疑框架，按以下顺序排查，99%的问题都能定位。

4.1 第一站：日志与输出

不要猜，先看日志。运行你的Agent时，确保日志输出是打开的。关注：

• on_goal_start打印了吗？目录找到了吗？文件列表正确吗？
• get_next_step被调用了多少次？每次返回的步骤描述对吗？
• on_step_result里记录的技能执行结果是成功还是失败？
• is_goal_complete是在什么时候返回True的？这个时机符合你的预期吗？

如果日志一片空白，或者在某条日志后戛然而止，那问题就出现在那条日志对应的环节。

4.2 第二站：Goal生命周期逻辑

这是高级Bug的高发区。对照检查：

• 初始化即完成/失败：在on_goal_start里，如果遇到边界情况（如无文件），你是否正确调用了mark_completed()？如果没有，get_next_step可能会返回None，而is_goal_complete可能永远不返回True，导致Agent空转或卡住。
• get_next_step返回None的时机：你的逻辑确保在处理完所有任务后才返回None吗？有没有可能因为某个条件判断错误提前返回了None？
• is_goal_complete的条件：这是重中之重。你的完成条件是否清晰、无歧义？是否依赖于某个可能永远无法达到的状态？用一个简单的测试验证：手动模拟任务流程，在每一步后检查这个函数的返回值是否符合预期。
• 技能执行异常处理：如果技能抛出一个未被捕获的异常，框架是会终止整个Goal，还是仅仅记录该步骤失败？你需要阅读框架文档，了解其错误处理机制，并在on_step_result中做好应对。

4.3 第三站：技能（Skill）本身

Agent停了，可能是因为某个技能执行时内部卡死或崩溃了。

• 超时：技能执行的操作（如网络请求、大文件读取、复杂计算）是否可能超时？框架或技能本身有没有设置超时机制？
• 资源耗尽：技能是否消耗了大量内存或CPU，导致进程被系统终止？
• 外部依赖：技能依赖的API、数据库、服务是否可用？认证信息是否有效？
• 输入输出：技能接收的参数格式对吗？它返回的结果是否符合on_step_result的预期？一个常见的错误是技能返回了非字典对象，导致后续处理出错。

调试建议：单独写一个小脚本，直接调用你的技能，传入各种边界情况的参数，看它是否都能稳定返回。

4.4 第四站：框架配置与版本

• 执行循环间隔：有些框架的agent.run()不是“紧循环”，可能有一个间隔（比如每秒检查一次）。如果间隔太长，可能会让你觉得Agent“停了”。查看配置。
• 记忆与上下文长度：对于复杂的、步骤多的Goal，LLM的上下文可能不够用，导致后续的规划或判断能力下降。检查框架是否提供了长上下文管理或摘要功能。
• 版本兼容性：你使用的Pi框架版本是否与你的代码示例兼容？API是否有变动？查阅CHANGELOG或版本迁移指南。
• 并发与异步：如果你的Goal涉及异步操作，确保你正确地处理了回调。不正确的异步代码可能导致执行流看似“停止”。

4.5 一个实用的调试技巧：添加“心跳”日志

在get_next_step方法里加一行特殊的日志，让你清晰地看到执行流。

def get_next_step(self): if self.current_file_index >= len(self.files_to_process): print(f"[DEBUG] get_next_step: 所有文件已处理，返回None。当前索引{self.current_file_index}, 总文件数{len(self.files_to_process)}") return None next_file = self.files_to_process[self.current_file_index] print(f"[DEBUG] get_next_step: 返回第{self.current_file_index+1}个文件: {next_file}") self.current_file_index += 1 return {...}

这样，如果日志显示[DEBUG] get_next_step: 所有文件已处理，返回None。之后，Agent没有结束，那问题一定出在is_goal_complete返回了False。反之，如果这条日志没出现Agent就停了，说明get_next_step提前返回了None，或者技能执行出了致命问题。

5. 超越基础：构建更健壮的持续执行Agent

理解了机制并解决了“停车”问题后，我们可以考虑如何让Agent更强大、更可靠。

5.1 引入状态持久化

上面的例子中，状态（current_file_index,results）都保存在内存里。如果Agent进程重启，所有进度都会丢失。对于长任务，需要将状态保存到外部（文件、数据库）。

• 在on_step_result中保存进度：每处理完一个文件，就将当前索引和结果写入一个状态文件。
• 在on_goal_start中读取进度：启动时检查是否存在状态文件，如果存在，则从中恢复current_file_index和results，实现“断点续跑”。

5.2 实现复杂的条件判断与动态规划

我们的例子是简单的顺序执行。真实的Goal可能需要动态规划。

• 分支逻辑：根据上一步的结果决定下一步做什么。例如，如果文件行数超过1000行，则调用“摘要”技能；否则，直接放入报告。
• 循环逻辑：直到满足某个条件才退出。例如，“持续监控日志文件，直到出现‘ERROR’关键词超过10次”。
• 这需要在get_next_step和is_goal_complete中实现更复杂的逻辑，可能还需要一个内部状态机来跟踪当前处于Goal的哪个阶段。

5.3 处理外部中断与用户交互

有时，用户可能想在中途暂停、修改任务或提供额外输入。

• 信号处理：让你的Agent能够捕获如Ctrl+C这样的中断信号，并优雅地保存状态后退出。
• 检查点：在get_next_step中定期检查是否有外部指令（如从一个消息队列中读取），从而动态调整计划。
• 这通常需要框架提供更高级的事件驱动或消息机制支持。

5.4 性能与资源监控

对于长时间运行的Agent，监控是必须的。

• 日志聚合：不要只打印到控制台，使用logging模块输出到文件，方便事后分析。
• 资源警报：在on_step_result中，可以检查任务执行时间。如果某个步骤异常耗时，记录警告。
• 限制机制：在get_next_step中，可以设置最大步骤数或最长运行时间，防止失控的无限循环。

6. 总结：从“会停”到“会跑”的关键思维转变

让Agent持续执行，不是一个魔法开关，而是一种系统设计。回顾全文，最关键的是完成以下思维转变：

1. 从“命令式”到“目标式”：你不要想着“先做A，再做B，然后做C”，而是告诉Agent“我的目标是X”，并设计一套机制（Goal）让它自己去拆解和完成X。
2. 从“单步调用”到“生命周期管理”：把Agent的一次执行看作一个拥有明确开始、步骤循环、状态判断、结束收尾的生命周期。你的代码是在管理这个周期。
3. 状态判断是灵魂：is_goal_complete这个函数是决定Agent何时停车的唯一裁判。它的逻辑必须绝对清晰、可靠。花最多的时间去设计和测试它。
4. 容错是保障：假设每一步都可能失败，并为失败设计处理路径（跳过、重试、记录）。一个因为一个步骤失败就整体崩溃的Agent是不可用的。
5. 日志是最好的调试器：在Goal和Skill的关键节点打入详细的、结构化的日志。当Agent表现异常时，日志是唯一能告诉你“它死前最后在想什么”的东西。

最后，不要试图在第一次就设计一个完美处理所有边界情况的Agent。我建议的实践路径是：先用最简单、最线性的任务（如我们的文件行数统计）跑通整个持续执行流程。然后，逐步引入一个复杂度（如错误处理），测试通过后，再引入下一个（如状态持久化）。每增加一个特性，都确保原有的核心流程依然稳固。这样，你构建的Agent才会既强大又可靠，真正成为能替你“跑完”任务的智能助手。