「Regnexe Python 实战系列」第 9 篇共 10 篇对应仓库examples/readme/09_cancel_and_resume.py。上一篇08. Agent 答案错了怎么排查先把 LLM 和工具调用事件打出来。一个产品里绕不开的需求只要你做过聊天式 Agent就一定会遇到“停止生成”按钮。很多实现只是把前端流断开用户看起来停止了但后端任务还在跑模型还在输出工具还在执行账单还在增加。regnexe-py 的acancel()解决的是后端任务取消按app_id / user_id / session_id找到正在运行的 Agent 任务并请求取消。实战代码先启动一个慢工具1、示例里故意做了一个慢工具plugin(idreports,nameReports Plugin)classReportsPlugin:agent_tool(Generate a financial report for a topic. Takes a while to run.,tags[reports])defgenerate_report(self,topic:str)-str:time.sleep(6)returnfReport on{topic}: revenue up 12% quarter-over-quarter.2、为了确定工具真的开始执行示例还写了一个监听器classToolStartListener(ConsoleEventListener):def__init__(self,tool_started:asyncio.Event)-None:super().__init__()self._tool_startedtool_startedasyncdefon_event(self,event_type:str,name:str,data:dict[str,Any])-None:awaitsuper().on_event(event_type,name,data)ifevent_typeTOOL_CALLED:self._tool_started.set()这个asyncio.Event用来避免靠sleep猜时机。3、构建 RegnexeAgent加入 ReportsPlugin 和 ToolStartListeneragent(RegnexeAgentBuilder().with_default_model(Vendor.DEEPSEEK,deepseek-v4-flash).with_plugin(ReportsPlugin()).with_event_listener(ToolStartListener(tool_started)).build())关键点ainvoke 要放到独立 Task 里执行取消要并发触达正在运行的任务所以示例这样启动runasyncio.create_task(agent.ainvoke(Generate a financial report on Q2 sales.,app_idreadme,user_idreader,session_idsession_id,))等工具开始后调用cancelledawaitagent.acancel(app_idreadme,user_idreader,session_idsession_id,)被取消的运行不会直接把异常抛给业务层而是返回AgentResult(statuscancelled)这对 API 层很友好。你可以把状态明确返回给前端而不是让一次取消变成 500 错误。继续执行同一个 session 再调用示例第二阶段没有调用特殊的resume()而是对同一个session_id发起普通ainvoke()result2awaitagent.ainvoke(Please go ahead and finish generating that report.,app_idreadme,user_idreader,session_idsession_id,)原因是这里没有人工审批卡点只是用户主动取消。LangGraph 已经在完成步骤后保存 checkpoint同一个 session 可以继续推进。如果是人工审批式中断那是另一个 APIaresume()。仓库顶层examples/10_interrupt_example.py专门演示这个场景。立即运行试试✈️ ✈️ ✈️ 小结acancel()适合这些产品交互聊天框停止生成用户取消长报告生成前端页面关闭后终止后端 Agent 任务任务耗时过长时主动停止记住两个关键点运行中的ainvoke()要放在独立asyncio.Task里取消要用同一组app_id / user_id / session_id定位任务到这里regnexe-py 的 9 个 README 示例就串完了从with_tool快速接入到 Skill/Sub-Agent、插件、目录加载、Marketplace、记忆、事件再到取消与继续。它们解决的不是“怎么调一次模型”而是“怎么把 Agent 做成一个能接业务系统的 Python 应用”。 上一篇08. Agent 答案错了怎么排查先把 LLM 和工具调用事件打出来 系列开篇00. 别再把 Python Agent 写成 Demo 了我开源了一个能管插件、记忆、事件、取消的框架 项目地址https://github.com/flower-trees/regnexe-py
「Regnexe Python 实战系列」第 9 篇共 10 篇对应仓库examples/readme/09_cancel_and_resume.py。上一篇08. Agent 答案错了怎么排查先把 LLM 和工具调用事件打出来。一个产品里绕不开的需求只要你做过聊天式 Agent就一定会遇到“停止生成”按钮。很多实现只是把前端流断开用户看起来停止了但后端任务还在跑模型还在输出工具还在执行账单还在增加。regnexe-py 的acancel()解决的是后端任务取消按app_id / user_id / session_id找到正在运行的 Agent 任务并请求取消。实战代码先启动一个慢工具1、示例里故意做了一个慢工具plugin(idreports,nameReports Plugin)classReportsPlugin:agent_tool(Generate a financial report for a topic. Takes a while to run.,tags[reports])defgenerate_report(self,topic:str)-str:time.sleep(6)returnfReport on{topic}: revenue up 12% quarter-over-quarter.2、为了确定工具真的开始执行示例还写了一个监听器classToolStartListener(ConsoleEventListener):def__init__(self,tool_started:asyncio.Event)-None:super().__init__()self._tool_startedtool_startedasyncdefon_event(self,event_type:str,name:str,data:dict[str,Any])-None:awaitsuper().on_event(event_type,name,data)ifevent_typeTOOL_CALLED:self._tool_started.set()这个asyncio.Event用来避免靠sleep猜时机。3、构建 RegnexeAgent加入 ReportsPlugin 和 ToolStartListeneragent(RegnexeAgentBuilder().with_default_model(Vendor.DEEPSEEK,deepseek-v4-flash).with_plugin(ReportsPlugin()).with_event_listener(ToolStartListener(tool_started)).build())关键点ainvoke 要放到独立 Task 里执行取消要并发触达正在运行的任务所以示例这样启动runasyncio.create_task(agent.ainvoke(Generate a financial report on Q2 sales.,app_idreadme,user_idreader,session_idsession_id,))等工具开始后调用cancelledawaitagent.acancel(app_idreadme,user_idreader,session_idsession_id,)被取消的运行不会直接把异常抛给业务层而是返回AgentResult(statuscancelled)这对 API 层很友好。你可以把状态明确返回给前端而不是让一次取消变成 500 错误。继续执行同一个 session 再调用示例第二阶段没有调用特殊的resume()而是对同一个session_id发起普通ainvoke()result2awaitagent.ainvoke(Please go ahead and finish generating that report.,app_idreadme,user_idreader,session_idsession_id,)原因是这里没有人工审批卡点只是用户主动取消。LangGraph 已经在完成步骤后保存 checkpoint同一个 session 可以继续推进。如果是人工审批式中断那是另一个 APIaresume()。仓库顶层examples/10_interrupt_example.py专门演示这个场景。立即运行试试✈️ ✈️ ✈️ 小结acancel()适合这些产品交互聊天框停止生成用户取消长报告生成前端页面关闭后终止后端 Agent 任务任务耗时过长时主动停止记住两个关键点运行中的ainvoke()要放在独立asyncio.Task里取消要用同一组app_id / user_id / session_id定位任务到这里regnexe-py 的 9 个 README 示例就串完了从with_tool快速接入到 Skill/Sub-Agent、插件、目录加载、Marketplace、记忆、事件再到取消与继续。它们解决的不是“怎么调一次模型”而是“怎么把 Agent 做成一个能接业务系统的 Python 应用”。 上一篇08. Agent 答案错了怎么排查先把 LLM 和工具调用事件打出来 系列开篇00. 别再把 Python Agent 写成 Demo 了我开源了一个能管插件、记忆、事件、取消的框架 项目地址https://github.com/flower-trees/regnexe-py
