agno 团队Team运行控制实战取消、后台执行、重试、模型继承与远程团队调用【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno导读在 agno 中Team 让多个 Agent 以路由、协同或团队模式协作完成复杂任务而运行控制Run Control决定了如何驾驭一次团队执行中途取消飞行中的运行、把长任务放到后台并轮询结果、对瞬时错误自动重试、让成员模型自动继承父团队模型以及通过 HTTP 调用部署在另一台 AgentOS 实例上的远程团队。本文以 cookbook/03_teams/14_run_control/README.md 为核心骨架结合仓库源码逐一拆解这七类运行控制能力读完后你将能直接照抄可运行的示例代码并在自己的团队应用中落地相同的控制逻辑。前置条件与运行方式按照 cookbook 目录的约定运行这些示例前需要准备通过direnv allow加载环境变量例如OPENAI_API_KEY保证模型调用可用使用.venvs/demo/bin/python来执行 cookbook 示例而非系统默认 Python部分示例依赖额外服务具体以文件 docstring 中标注为准例如 PostgreSQLlocalhost:5532或远程 AgentOS 实例:7778。标准运行命令如下.venvs/demo/bin/python cookbook/03_teams/14_run_control/cancel_run.py .venvs/demo/bin/python cookbook/03_teams/14_run_control/model_inheritance.py .venvs/demo/bin/python cookbook/03_teams/14_run_control/retries.py本文所涉 8 个示例的速查表对应原文档 Examples 表示例演示内容依赖cancel_run.py从独立线程取消一次飞行中的团队运行—cancel_run_persistence.py流式中途取消并验证部分内容与消息已持久化到数据库PostgreSQLteam_cancel_while_member_runs.py成员 Agent 正在流式输出时取消团队运行取消向下传播团队与成员运行均以cancelled落库PostgreSQLbackground_execution.py启动后立即返回PENDING的后台运行轮询完成或取消PostgreSQLbackground_execution_metrics.py团队后台运行同样完整记录指标可从落库的 run 上读回PostgreSQLmodel_inheritance.py成员模型从父团队模型继承—remote_team.py调用并流式读取托管在远程 AgentOS 实例上的团队AgentOS 运行于:7778retries.py团队针对瞬时运行错误的重试配置—其中 7 个示例在进程内驱动团队对象只有remote_team.py走 HTTP 跨进程。暂停团队等待人工决策属于人类在环Human-in-the-Loop范畴见 cookbook/03_teams/20_human_in_the_loop/通过 AgentOS 路由驱动的同一套运行控制机制见 cookbook/05_agent_os/04_run_lifecycle/。一、取消飞行中的团队运行cancel_run.pycancel_run.py演示了最经典的取消场景团队正在流式输出长篇内容时从另一个线程调用team.cancel_run(run_id)把运行标记为取消。实现拆解示例构造了一个包含 StorytellerAgent 与 EditorAgent 的写作团队让它在streamTrue模式下写一篇 2000 字以上的故事同时用两个线程并行执行long_running_task运行线程遍历team.run(...)产生的流式 chunk收集TeamRunEvent.run_content/RunEvent.run_content事件中的内容片段并监听run_cancelled事件cancel_after_delay取消线程睡眠若干秒后取出运行时捕获的run_id调用team.cancel_run(run_id)。核心取消调用只有一行success team.cancel_run(run_id)在源码层面Team.cancel_run是一个静态方法直接委托给运行管理器staticmethod def cancel_run(run_id: str) - bool: return _run.cancel_run(run_idrun_id) staticmethod async def acancel_run(run_id: str) - bool: return await _run.acancel_run(run_idrun_id)见 libs/agno/agno/team/team.py。它返回布尔值表示取消是否成功若运行不存在或已结束返回False因此示例中对取消失败分支做了提示处理。事件处理要点流式事件中需要同时关注两个层级的取消事件因为成员运行与团队运行都可能被取消RunEvent.run_cancelled某个成员 Agent 的运行被取消TeamRunEvent.run_cancelled整个团队运行被取消。示例在收到任一取消事件后立即返回并把运行状态、run_id、是否取消以及取消前已收集的部分内容写入run_id_container最后统一打印结果。运行状态使用RunStatus枚举见 libs/agno/agno/run/base.py判断是否正常完成其取值包括PENDING、COMPLETED、CANCELLED、ERROR等。二、取消 持久化验证部分内容被保留cancel_run_persistence.py现实场景中取消不是一键清空已产出的部分内容与消息应当被保存供会话恢复或审计。cancel_run_persistence.py演示了这一点要求 PostgreSQL 运行在localhost:5532可参考 cookbook/scripts/run_pgvector.sh。关键配置team Team( nameResearchTeam, members[researcher], modelOpenAIResponses(idgpt-5.4), dbPostgresDb(db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai), store_tool_messagesTrue, store_history_messagesTrue, )要点db团队运行的会话与运行记录写入 PostgreSQLstore_tool_messagesTrue工具调用消息同样落库store_history_messagesTrue历史消息落库保证取消后会话上下文不丢失。取消与验证流程for event in team.run(input..., streamTrue, stream_eventsTrue): if run_id is None and hasattr(event, run_id) and event.run_id: run_id event.run_id if hasattr(event, content) and event.content: content_chunks.append(event.content) print(event.content, end, flushTrue) # 收集到 20 个内容块后主动取消 if len(content_chunks) 20 and run_id and not cancelled: team.cancel_run(run_id) cancelled True if hasattr(event, event) and event.event TeamRunEvent.run_cancelled: print(\nRun was cancelled) break注意这里使用了stream_eventsTrue事件对象同时携带run_id、content、event等属性因此在遍历时可直接判断。取消后通过会话对象验证持久化结果session team.get_session(session_idteam.session_id) if session and session.runs: last_run session.runs[-1] print(fStatus: {last_run.status}) print(fContent length: {len(last_run.content or )}) print(fMessages: {len(last_run.messages or [])})由此验证即便运行被标记为CANCELLED其部分 content 与消息列表依然完整存在于数据库可作为后续继续会话的基础。三、取消传播成员正在流式输出时取消团队运行team_cancel_while_member_runs.py覆盖更复杂的情形团队已把任务委托给成员 Agent路由模式成员正在流式输出长文时取消整个团队运行。取消需要从团队传播到飞行中的成员并且两条运行记录都以cancelled状态落库。核心配置from agno.run.team import RunCancelledEvent as TeamRunCancelledEvent from agno.run.team import ToolCallStartedEvent from agno.team.mode import TeamMode team Team( nameCancelWhileMemberRuns, modeTeamMode.route, modelOpenAIResponses(idgpt-5.4), members[researcher], dbPostgresDb(db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai), show_members_responsesTrue, store_tool_messagesTrue, store_history_messagesTrue, )要点modeTeamMode.route路由模式团队通过工具调用把任务委派给成员show_members_responsesTrue在团队输出中显示成员的响应内容事件类型换成了更细粒度的ToolCallStartedEvent与RunCancelledEvent。捕捉委派时机示例监听ToolCallStartedEvent当工具名是delegate_task_to_member时判定委派已开始if isinstance(event, ToolCallStartedEvent): tool_name getattr(getattr(event, tool, None), tool_name, None) if tool_name delegate_task_to_member: delegation_started True随后一旦成员内容块达到 10 个就调用team.cancel_run(run_id)使取消精确落在成员飞行途中。验证双向持久化取消后会话中同时存在团队运行与成员运行两类记录区分依据是team_id与agent_idteam_runs [run for run in session.runs if getattr(run, team_id, None)] member_runs [run for run in session.runs if getattr(run, agent_id, None)]成员运行通过parent_run_id指回父级团队运行打印时可见run.agent_name、run.parent_run_id、run.status等字段。这正是取消传播机制的落库证据团队运行与成员运行都会持久化为cancelled形成完整的父子运行链。四、后台执行启动即返回、轮询与异步取消background_execution.py演示团队的后台执行模式调用team.arun(..., backgroundTrue)后立即返回运行状态为PENDING实际工作在后台继续随后用team.aget_run_output轮询直到COMPLETED或ERROR也可以随时用team.acancel_run取消。启动后台运行db PostgresDb( db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai, session_tableteam_bg_exec_sessions, # 自定义会话表名 ) run_output await team.arun( What are the three laws of thermodynamics? Summarize each in one sentence., backgroundTrue, ) print(fRun ID: {run_output.run_id}) print(fSession ID: {run_output.session_id}) print(fStatus: {run_output.status}) assert run_output.status RunStatus.pending # 立即返回状态为 PENDING后台运行依赖数据库因为运行结果必须能被后续轮询读回因此PostgresDb通过session_table参数可自定义表名以隔离不同用例的数据。轮询完成for i in range(60): await asyncio.sleep(1) result await team.aget_run_output( run_idrun_output.run_id, session_idrun_output.session_id, ) if result is None: continue if result.status RunStatus.completed: print(fCompleted! Content:\n{result.content}) break elif result.status RunStatus.error: print(fFailed! Content: {result.content}) break对应同步/异步 API 在 libs/agno/agno/team/team.py 中均有定义get_run_output与aget_run_output。取消后台运行await asyncio.sleep(3) cancelled await team.acancel_run(run_idrun_output.run_id) print(fCancel result: {cancelled}) await asyncio.sleep(1) result await team.aget_run_output(run_idrun_output.run_id, session_idrun_output.session_id) if result: print(fFinal status: {result.status})即启动即返回 → 轮询 → 完成或取消的完整闭环适合耗时任务、队列式作业等不阻塞调用方的场景。五、后台运行的指标可观测性background_execution_metrics.pybackground_execution_metrics.py进一步证明团队后台运行完成后指标与前台同步运行完全一致包括 token 统计、模型详情、耗时以及成员级拆分且都能从落库的 run 上读回。示例使用带工具的成员stock_searcher Agent( nameStock Searcher, modelOpenAIChat(idgpt-5.6-luna), roleSearches for stock information., tools[YFinanceTools(enable_stock_priceTrue)], ) team Team( nameStock Research Team, modelOpenAIChat(idgpt-5.6-luna), members[stock_searcher], dbdb, show_members_responsesTrue, store_member_responsesTrue, )后台运行完成后示例依次打印四层指标团队级指标result.metrics用rich.pretty.pprint完整展示模型详情拆分result.metrics.details按模型类型分组列出成员级指标遍历result.member_responses逐个打印member_response.metrics需要store_member_responsesTrue才会保留成员响应会话级指标team.get_session_metrics()返回SessionMetrics对应源码 libs/agno/agno/team/team.py。这说明后台执行并非黑盒成本、耗时、成员贡献等观测数据可以完整回溯方便后续计费与质量分析。六、模型继承成员自动继承父团队模型model_inheritance.py演示一个重要的默认行为——成员模型从父团队模型继承。当成员 Agent 未显式指定model时Team 会把自己的模型下发给成员成员也可显式覆盖子团队同理。配置组合researcher Agent(nameResearcher, roleResearch and gather information, instructions[...]) writer Agent(nameWriter, roleWrite content based on research, ...) editor Agent( nameEditor, roleEdit and refine content, modelOpenAIResponses(idgpt-5.2), # 显式指定覆盖继承 ... ) analyst Agent(nameAnalyst, roleAnalyze data and provide insights) sub_team Team( nameAnalysis Team, modelOpenAIResponses(idgpt-5.2), members[analyst], ) team Team( nameContent Production Team, modelOpenAIResponses(idgpt-5.2), members[researcher, writer, editor, sub_team], instructions[...], )继承效果验证运行前调用team.initialize_team()源码见 libs/agno/agno/team/team.py完成成员初始化与模型下发然后打印各成员最终生效的模型team.initialize_team() print(fResearcher model: {researcher.model.id}) # 继承 team 的 gpt-5.2 print(fWriter model: {writer.model.id}) # 继承 team 的 gpt-5.2 print(fEditor model: {editor.model.id}) # 显式指定保持不变 print(fAnalyst model: {analyst.model.id}) # 从子团队继承由此可以看出researcher、writer未声明模型最终拿到团队模型的 ideditor因显式声明而保持自己的模型analyst作为子团队的成员继承的是其直接父级Analysis Team的模型。这一机制极大简化了大型团队的模型统一管理——只需在根团队配置一次模型即可统一下发。七、远程团队通过 HTTP 调用另一台 AgentOS 上的团队remote_team.py是 8 个示例中唯一走网络协议的它用RemoteTeam调用并流式读取托管在远程 AgentOS 实例上的团队属于 AgentOS 跨进程协作的一部分。客户端配置from agno.team import RemoteTeam remote_team RemoteTeam( base_urlhttp://localhost:7778, # 远程 AgentOS 实例地址 team_idresearch-team, # 远程实例上的团队 ID )在源码层面libs/agno/agno/team/remote.pyRemoteTeam继承BaseRemote构造参数为base_url远程实例地址、timeout、protocol、a2a_protocol、config_ttl并持有team_id用于从远程实例拉取团队配置与运行任务。普通调用与流式调用# 普通调用 response await remote_team.arun( What is the capital of France?, user_iduser-123, session_idsession-456, ) print(response.content) # 流式调用 async for chunk in remote_team.arun( Tell me a 2 sentence horror story, session_idsession-456, user_iduser-123, streamTrue, ): if hasattr(chunk, content) and chunk.content: print(chunk.content, end, flushTrue)服务不可用的优雅降级示例对远程连接失败做了显式兜底捕获连接类异常并给出可操作提示except (ConnectionError, TimeoutError, OSError, socket.gaierror, RemoteServerUnavailableError) as exc: print(\nRemoteTeam server is not available. Start a remote AgentOS instance at http://localhost:7778 and rerun this cookbook.) print(fOriginal error: {exc})RemoteServerUnavailableError来自agno.exceptions用于标识远程服务器不可达。这提醒读者远程团队调用属于分布式场景必须把网络异常当作一等公民处理。八、重试为瞬时错误配置团队级重试retries.py演示团队的运行级重试配置。对于网络抖动、限流等瞬时错误与其让整次运行失败不如在团队层面自动重试。配置示例team Team( members[sarah, mike], # sarah 带 WebSearchToolsmike 负责写摘要 retries3, # 最多重试 3 次 delay_between_retries1, # 每次重试间隔 1 秒 exponential_backoffTrue,# 启用指数退避 )三个关键参数的组合效果retries运行失败后的最大重试次数控制容错上限delay_between_retries相邻两次重试的基础间隔秒exponential_backoff开启后每次重试的等待时间按指数增长避免集中重试加剧服务压力。这种固定重试次数 基础间隔 指数退避的组合是生产级系统的通用实践尤其适合成员带外部搜索工具如WebSearchTools这类容易触发限流的场景。总结如何为你的团队选择合适的运行控制策略综合原文档与 8 个示例可以在实际项目中按需组合需要及时止损长任务参考cancel_run.py从独立线程调用team.cancel_run(run_id)并用RunEvent.run_cancelled/TeamRunEvent.run_cancelled区分成员级与团队级取消取消后要保留上下文开启store_tool_messages、store_history_messages并配置db参考cancel_run_persistence.py验证部分内容与消息落库路由模式下取消要向下传播参考team_cancel_while_member_runs.py关注ToolCallStartedEvent中的delegate_task_to_member委派时机并用team_id/agent_id区分团队与成员运行记录不想阻塞调用方使用arun(..., backgroundTrue)aget_run_output轮询 acancel_run取消参考background_execution.py与background_execution_metrics.py后者还能读取团队、模型、成员、会话四层指标统一团队模型口径成员不指定模型时自动继承父团队模型显式指定则覆盖参考model_inheritance.py跨进程调用团队用RemoteTeam(base_url..., team_id...)并像remote_team.py一样处理网络异常提升瞬时错误下的稳定性组合retries、delay_between_retries、exponential_backoff参考retries.py。需要人工决策的暂停属于 Human-in-the-Loop 范畴cookbook/03_teams/20_human_in_the_loop/而同一套运行控制若要通过 AgentOS 路由驱动可继续研读 cookbook/05_agent_os/04_run_lifecycle/。所有示例源码均可直接运行验证是学习 agno 团队运行控制机制的最佳入口。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考