Agno Agent 高级实战指南缓存、压缩、并发、事件、重试、调试与序列化【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇指南聚焦 Agnoagno仓库中 cookbook/02_agents/14_advanced 这一高级专题目录系统讲解 Agent 在生产级场景下的七大进阶能力模型响应缓存、上下文压缩、并发与后台执行、运行取消与持久化、生命周期事件监听、重试与调试日志、以及 Agent 序列化。读完本文你将能够基于仓库中的可运行示例直接为自己的 Agent 应用叠加这些工程能力并理解其底层实现位置。目录概览与文件清单14_advanced目录围绕生产可用展开覆盖缓存、压缩、并发、事件、重试、调试与序列化七大主题。以下是目录内全部示例文件的职责说明文件主题核心能力cache_model_response.py缓存缓存模型响应避免重复 API 调用advanced_compression.py压缩基于 token 上限的工具调用上下文压缩tool_call_compression.py压缩压缩工具调用结果控制上下文体积compression_events.py压缩/事件压缩过程中的事件流监听concurrent_execution.py并发用asyncio.gather并发执行多个任务background_execution.py后台后台运行 轮询 取消background_execution_structured.py后台带结构化输出的后台执行background_execution_concurrency.py后台/并发后台并发执行background_execution_metrics.py后台/指标后台运行的指标采集background_streaming_resume.py后台/流式后台流式执行与断点恢复redis_event_stream_resume.py事件/恢复基于 Redis 事件流的恢复cancel_run.py取消取消正在运行的 Agentcustom_cancellation_manager.py取消自定义取消逻辑agent_run_cancel_persistence.py取消/持久化取消运行并验证部分内容落库basic_agent_events.py事件监听 Agent 生命周期事件reasoning_agent_events.py事件推理步骤中的事件retries.py重试带指数退避的重试配置debug.py调试开启 verbose 调试输出custom_logging.py日志自定义日志配置metrics.py/multi_model_metrics.py/session_metrics.py/session_summary_metrics.py/streaming_metrics.py/tool_call_metrics.py指标访问 Agent 运行指标agent_serialization.py序列化Agent 的序列化与反序列化interchange_model/模型互操作跨提供商的模型互换OpenAI/Claude/Gemini 等运行环境与前置条件在运行本目录任何示例之前需要完成以下准备加载环境变量在仓库根目录执行direnv allow确保OPENAI_API_KEY等密钥可用。创建演示环境运行./scripts/demo_setup.sh创建虚拟环境之后所有 cookbook 均通过.venvs/demo/bin/python执行。可选依赖部分示例依赖本地服务或特定提供商的 API 密钥。例如agent_run_cancel_persistence.py与background_execution.py要求本地 PostgreSQLpostgresqlpsycopg://ai:ailocalhost:5532/ai可参照仓库内cookbook/scripts/run_pgvector.sh启动。运行单个示例的统一命令格式为.venvs/demo/bin/python cookbook/02_agents/14_advanced/file.py例如.venvs/demo/bin/python cookbook/02_agents/14_advanced/cache_model_response.py模型响应缓存消除重复 API 调用cache_model_response.py演示了 Agno 最直接的优化手段对相同请求复用模型响应从而减少 API 调用与延迟。from agno.agent import Agent from agno.models.openai import OpenAIResponses # 在模型层开启响应缓存 agent Agent(modelOpenAIResponses(idgpt-5.6-luna, cache_responseTrue)) # 同一查询执行两次第一次为 Cache Miss第二次为 Cache Hit for i in range(1, 3): response agent.run( Write me a short story about a cat that can talk and solve problems. ) print(response.content) print(f\n Elapsed time: {response.metrics.duration:.3f}s)关键点缓存开关位于模型层通过OpenAIResponses(..., cache_responseTrue)开启而不是在 Agent 上配置。这是理解缓存粒度的关键——缓存与具体模型实例绑定。验证方式示例连续两次运行相同 query并在输出中标注 Cache Miss (First Request) 与 Cache Hit (Cached Response)同时通过response.metrics.duration对比两次耗时。缓存命中时耗时通常显著下降。注意缓存语义依赖具体模型提供商的实现与密钥配置示例中的gpt-5.6-luna为仓库当前示例所用模型 ID实际使用时应替换为你环境内可用的模型。上下文压缩控制长会话与工具调用的 token 消耗上下文窗口是 Agent 应用的硬约束。14_advanced提供了三级压缩方案核心入口是 libs/agno/agno/compression 模块中的CompressionManager。基于 token 上限的工具调用压缩advanced_compression.py该示例为竞品情报分析师场景定制了压缩策略当上下文达到指定 token 阈值时对工具调用历史进行压缩。from agno.agent import Agent from agno.compression.manager import CompressionManager from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.tools.websearch import WebSearchTools compression_prompt You are a compression expert. Your goal is to compress web search results for a competitive intelligence analyst. MUST PRESERVE: - Competitor names and specific actions (product launches, partnerships, acquisitions, pricing changes) - Exact numbers (revenue, market share, growth rates, pricing, headcount) - Precise dates (announcement dates, launch dates, deal dates) - Direct quotes from executives or official statements - Funding rounds and valuations MUST REMOVE: - Company history and background information - General industry trends (unless competitor-specific) - Analyst opinions and speculation (keep only facts) ... OUTPUT FORMAT: Return a bullet-point list where each line follows this format: [Company Name] - [Date]: [Action/Event] ([Key Numbers/Details]) Keep it under 200 words total. Be ruthlessly concise. Facts only. compression_manager CompressionManager( modelOpenAIResponses(idgpt-5-mini), compress_token_limit5000, # 上下文 token 超过该阈值即触发压缩 compress_tool_call_instructionscompression_prompt, # 领域定制的压缩指令 ) agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[WebSearchTools()], descriptionSpecialized in tracking competitor activities, instructionsUse the search tools and always use the latest information and data., dbSqliteDb(db_filetmp/token_based_tool_call_compression.db), compression_managercompression_manager, add_history_to_contextTrue, # 将历史加入上下文 num_history_runs3, # 保留最近 3 轮历史 session_idtoken_based_tool_call_compression, )参数要点compress_token_limit触发压缩的上下文 token 阈值此处为 5000。这是基于 token 的压缩的核心区别于按轮次/数量压缩。compress_tool_call_instructions领域定制提示词示例通过MUST PRESERVE保留竞品名称、精确数字、日期、高管原话、融资轮次与MUST REMOVE剔除公司历史、行业泛论、分析师观点、营销话术来确保压缩后信息无损于业务目标。这展示了压缩指令模板化的最佳实践。配套存储示例使用SqliteDb持久化会话配合add_history_to_contextTrue与num_history_runs3使压缩与多轮历史协同工作。工具调用结果压缩tool_call_compression.pytool_call_compression.py聚焦单次工具调用结果体积过大的问题——例如 Web 搜索返回大量页面摘要时可在写入上下文前对工具结果进行压缩。它与advanced_compression.py的区别在于粒度前者压缩历史的工具调用后者针对即将进入上下文的工具结果。压缩过程事件compression_events.pycompression_events.py演示在压缩发生时订阅事件流用于观察压缩的触发时机、被压缩的内容量以及压缩后的 token 节省情况。这在调优compress_token_limit时非常有用——你可以据此判断阈值是否过松频繁压缩或过紧上下文仍超限。从源码结构看压缩逻辑统一由 libs/agno/agno/compression/manager.py 中的CompressionManager调度其核心职责是监测上下文 token 用量 → 达到阈值时调用压缩模型 → 用压缩结果替换原始工具调用历史 → 触发压缩事件。compress_token_limit与压缩指令均作为该管理器构造参数传入。并发执行单个 Agent 实例并行处理多任务concurrent_execution.py演示了通过asyncio.gather让同一个 Agent 实例并发执行多个独立研究任务import asyncio from agno.agent import Agent from agno.models.openai import OpenAIResponses from agno.tools.duckduckgo import DuckDuckGoTools providers [openai, anthropic, ollama, cohere, google] instructions Your task is to write a well researched report on AI providers. The report should be unbiased and factual. # 在循环外只创建一次 Agent —— 这是正确的并发模式 agent Agent( modelOpenAIResponses(idgpt-5-mini), instructionsinstructions, tools[DuckDuckGoTools()], ) async def get_reports(): 使用同一个 Agent 实例并发运行多个研究任务。 tasks [ agent.arun(fWrite a report on the following AI provider: {provider}) for provider in providers ] results await asyncio.gather(*tasks) return results async def main(): results await get_reports() for result in results: pprint(result.content) if __name__ __main__: asyncio.run(main())实践要点复用实例而非复制实例示例注释明确指出Create the agent ONCE outside the loop - this is the correct pattern。并发安全的前提是共享同一个Agent对象调用其异步接口arun。异步 APIagent.arun(...)返回协程由asyncio.gather统一调度5 个研究任务并行推进。适用边界并发任务之间应相互独立无共享可变状态且底层模型 API 需支持并发请求配额。后台执行提交即返回轮询或取消background_execution.py展示了后台执行的完整生命周期提交backgroundTrue的 run 后立即返回PENDING状态实际工作在后台推进调用方随后轮询结果或随时取消。示例一后台运行 轮询db PostgresDb( db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai, session_tablebackground_exec_sessions, ) agent Agent( nameBackgroundAgent, modelOpenAIResponses(idgpt-5-mini), descriptionAn agent that runs in the background, dbdb, ) # 后台启动 —— 立即返回 PENDING run_output await agent.arun( What is the capital of France? Answer in one sentence., backgroundTrue, ) assert run_output.status RunStatus.pending # 每秒轮询一次直至 completed / error for i in range(30): await asyncio.sleep(1) result await agent.aget_run_output( run_idrun_output.run_id, session_idrun_output.session_id, ) if result.status RunStatus.completed: print(f\nCompleted! Content: {result.content}) break elif result.status RunStatus.error: break示例二与示例三取消后台运行与先取消后启动# 取消一个正在运行的后台任务 cancelled await agent.acancel_run(run_idrun_output.run_id) # 取消-先于-启动语义预生成 run_id在启动前取消 from agno.run.cancel import cancel_run from uuid import uuid4 run_id str(uuid4()) cancel_run(run_id) # 先取消 run_output await agent.arun( This should be cancelled before it runs., backgroundTrue, run_idrun_id, # 启动时检测到已取消 )关键机制状态机后台运行的状态由agno.run.base.RunStatus定义pending→completed/error/cancelled。提交后立即返回pending随后在数据库中更新。轮询接口aget_run_output(run_id, session_id)按 run_id session_id 从数据库读取最新状态。取消语义acancel_run取消进行中的运行agno.run.cancel.cancel_run(run_id)支持先取消后启动——即便 run 尚未开始也可以在启动时被检测到并取消。这是分布式/长时间运行场景下的重要容错能力。存储前提后台执行依赖数据库示例用 PostgreSQL记录 run 状态无存储时该能力不可用。从源码结构看后台执行的调度与状态查询实现在 libs/agno/agno/run 目录下background_execution.py是backgroundTrue参数与运行状态存储之间的完整调用链示例。目录中还有background_execution_structured.py后台 Pydantic 结构化输出、background_execution_concurrency.py后台并发、background_execution_metrics.py后台指标、background_streaming_resume.py后台流式 恢复以及redis_event_stream_resume.py基于 Redis 事件流恢复运行可按需组合上述模式。运行取消中断长任务与部分内容持久化基础取消cancel_run.pycancel_run.py通过双线程演示取消流程一个线程运行长任务写 2000 字以上的长故事并流式消费内容另一个线程在延迟后调用agent.cancel_run(run_id)for chunk in agent.run(...长故事提示词..., streamTrue): if run_id not in run_id_container and chunk.run_id: run_id_container[run_id] chunk.run_id if chunk.event RunEvent.run_content: content_pieces.append(chunk.content) print(chunk.content, end, flushTrue) # 运行被取消时会发出 RunEvent.run_cancelled 事件 elif chunk.event RunEvent.run_cancelled: print(f\n[CANCELLED] Run was cancelled: {chunk.run_id}) return elif hasattr(chunk, status) and chunk.status RunStatus.completed: final_response chunk # 取消线程 success agent.cancel_run(run_id)要点取消是标记式的cancel_run(run_id)返回布尔值表示取消是否被受理对于已结束或不存在的 run返回失败并给出提示。事件驱动感知取消发生后流中会出现RunEvent.run_cancelled事件业务代码据此收尾示例中保存已生成的部分内容。run_id 传递多线程场景下需通过共享容器示例为run_id_containerdict在运行线程与取消线程间传递run_id。取消 持久化验证agent_run_cancel_persistence.py该示例将取消能力与数据库持久化结合流式运行中收集到 20 个内容块后主动取消随后从会话中取回最后一次 run验证部分内容与消息是否已写入数据库agent Agent( nameStoryteller, modelOpenAIResponses(idgpt-5.4), instructionsYou are a storyteller. Write very long detailed stories., dbPostgresDb(db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai), store_tool_messagesTrue, store_history_messagesTrue, ) for event in agent.run(input..., streamTrue, stream_eventsTrue): if len(content_chunks) 20 and run_id and not cancelled: agent.cancel_run(run_id) cancelled True if event.event RunEvent.run_cancelled: print(\nRun was cancelled) break # 验证从会话中读取最后一条 run 的部分内容与消息 session agent.get_session(session_idagent.session_id) 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 [])})这证明了 Agno 的一个重要设计取消不等于丢失。通过store_tool_messages与store_history_messages开启消息存储后中断的 run 的已生成内容、状态与消息都会被持久化可支撑后续断点续跑或人工审查。custom_cancellation_manager.py则进一步允许替换取消逻辑的实现自定义CancellationManager用于接入自定义的分布式取消信号如 Redis 发布订阅等场景。事件监听洞察 Agent 生命周期basic_agent_events.py展示了如何通过stream_eventsTrue订阅 Agent 的完整生命周期事件这是构建可观测性、日志与 UI 流式展示的基础from agno.agent import RunEvent from agno.agent.agent import Agent finance_agent Agent( idfinance-agent, nameFinance Agent, modelOpenAIResponses(idgpt-5.2), tools[YFinanceTools()], ) async def run_agent_with_events(prompt: str): content_started False async for run_output_event in finance_agent.arun( prompt, streamTrue, stream_eventsTrue ): if run_output_event.event in [RunEvent.run_started, RunEvent.run_completed]: print(f\nEVENT: {run_output_event.event}) if run_output_event.event in [RunEvent.tool_call_started]: print(fTOOL CALL: {run_output_event.tool.tool_name}) print(fTOOL CALL ARGS: {run_output_event.tool.tool_args}) if run_output_event.event in [RunEvent.tool_call_completed]: print(fTOOL CALL: {run_output_event.tool.tool_name}) print(fTOOL CALL RESULT: {run_output_event.tool.result}) if run_output_event.event in [RunEvent.run_content]: print(run_output_event.content, end)事件体系要点事件枚举核心生命周期事件定义在agno.run.agent.RunEvent本示例从agno.agent导入RunEvent覆盖run_started、run_completed、tool_call_started、tool_call_completed、run_content、run_cancelled等。取消示例中使用的RunEvent.run_cancelled与压缩示例中的压缩事件同属该体系。事件负载工具调用事件携带tool对象tool_name、tool_args、result内容事件携带content流式片段每个事件还携带run_id取消/持久化示例正是利用这一点拿到 run_id。推理事件reasoning_agent_events.py补充了推理步骤reasoning steps期间的事件适用于需要展示模型思考过程的可解释性场景。重试指数退避提升鲁棒性retries.py演示了网络抖动、限流等临时错误下的自动重试agent Agent( nameWeb Search Agent, roleSearch the web for information, tools[WebSearchTools()], retries3, # 出错时整个 Agent run 最多重试 3 次 delay_between_retries1, # 每次重试之间的基础延迟秒 exponential_backoffTrue, # 开启后延迟每次翻倍 )参数语义retriesAgent run 在出错情况下的最大重试次数示例为 3。delay_between_retries基础延迟秒数。exponential_backoff为 True 时每次重试的延迟翻倍1s → 2s → 4s这是对 API 限流类错误最友好的退避策略避免雪崩式重试。调试与自定义日志开启调试模式debug.py# 方式一Agent 级全局开启所有 run 输出更详细的日志 agent Agent(modelOpenAIResponses(idgpt-5-mini), debug_modeTrue) agent.print_response(inputTell me a joke.) # 方式二单次 run 临时开启 agent Agent(modelOpenAIResponses(idgpt-5-mini)) agent.print_response(inputTell me a joke., debug_modeTrue)debug_modeTrue会产生更 verbose 的输出请求/响应细节、内部调用链适合排查 Agent 行为异常。注意单次 run 的debug_mode参数优先级高于 Agent 级配置可在不污染全局配置的前提下做一次性的深度排查。自定义日志custom_logging.pyAgno 的日志统一由 libs/agno/agno/utils/log.py 提供可通过configure_agno_logging无缝替换为自有 loggerimport logging from agno.utils.log import configure_agno_logging, log_info def get_custom_logger(): custom_logger logging.getLogger(custom_logger) handler logging.StreamHandler() formatter logging.Formatter([CUSTOM_LOGGER] %(levelname)s: %(message)s) handler.setFormatter(formatter) custom_logger.addHandler(handler) custom_logger.setLevel(logging.INFO) custom_logger.propagate False return custom_logger custom_logger get_custom_logger() # 全局替换此后 Agno 的所有日志均走自定义 logger configure_agno_logging(custom_default_loggercustom_logger) log_info(This is using our custom logger!) agent Agent() agent.print_response(What can I do to improve my sleep?)要点configure_agno_logging(custom_default_logger...)是一次性全局配置之后agno.utils.log中所有日志函数如log_info以及 Agent 内部日志都会路由到自定义 logger。这是将 Agno 接入企业日志体系JSON 结构化、日志采集、级别策略的标准入口。Agent 序列化保存、加载与重建agent_serialization.py演示了 Agent 的三种物化方式字典互转、按版本持久化到数据库、从数据库加载from agno.agent import Agent from agno.db.sqlite import SqliteDb agent_db SqliteDb(db_filetmp/agents.db) agent Agent( idserialization-demo-agent, nameSerialization Demo Agent, modelOpenAIResponses(idgpt-5.2), dbagent_db, ) # 方式一与 dict 互转内存内重建 config agent.to_dict() recreated Agent.from_dict(config) # 方式二按版本保存到数据库再按 id version 加载 version agent.save() loaded Agent.load(idagent.id, dbagent_db, versionversion) recreated.print_response(Say hello from a recreated agent., streamTrue) loaded.print_response(Say hello from a loaded agent., streamTrue)三种途径的适用场景to_dict()/Agent.from_dict(config)轻量重建适合配置迁移、复制 Agent 模板或在不依赖数据库的环境间传递 Agent 定义。agent.save()→version将 Agent 序列化后写入数据库此处为SqliteDb数据库无关并返回版本号。Agent.load(id..., db..., version...)按唯一 id 与版本号还原历史 Agent 配置适合版本管理与审计。序列化能力由 libs/agno/agno/agent 中的 Agent 定义支撑结合任意 Agno 支持的数据库SQLite/PostgreSQL 等即可获得配置即数据的持久化体验。运行指标量化 Agent 行为目录下metrics.py、multi_model_metrics.py、session_metrics.py、session_summary_metrics.py、streaming_metrics.py、tool_call_metrics.py及background_execution_metrics.py构成一套完整的指标观察矩阵分别从以下维度量化运行单次 run 指标metrics.py通过response.metrics.duration等字段获取耗时缓存示例已用到。流式指标streaming_metrics.py衡量首 token 延迟与整体流式传输耗时。工具调用指标tool_call_metrics.py统计各工具调用的次数与耗时定位瓶颈工具。会话与汇总指标session_metrics.py、session_summary_metrics.py跨多次 run 聚合评估长会话的整体成本与表现。多模型对比multi_model_metrics.py在同一任务上横向对比不同模型的耗时与输出服务于选型。后台运行指标background_execution_metrics.py后台任务同样暴露指标便于无人值守场景的监控采集。这些指标对象可直接对接自建监控看板或日志采集管线与上文的事件体系互补——事件描述发生了什么指标量化花了多少成本/时间。小结七大高级能力速查能力关键 API / 参数代表示例模型响应缓存OpenAIResponses(cache_responseTrue)cache_model_response.py上下文压缩CompressionManager(compress_token_limit, compress_tool_call_instructions)advanced_compression.py并发执行agent.arun(...)asyncio.gatherconcurrent_execution.py后台执行arun(backgroundTrue)aget_run_output轮询 acancel_runbackground_execution.py运行取消agent.cancel_run(run_id)RunEvent.run_cancelledcancel_run.py、agent_run_cancel_persistence.py事件监听stream_eventsTrueRunEvent.*basic_agent_events.py重试retries、delay_between_retries、exponential_backoffretries.py调试/日志debug_modeTrue、configure_agno_logging(custom_default_logger...)debug.py、custom_logging.py序列化to_dict()/from_dict()、save()/load()agent_serialization.py指标response.metrics、各专项 metrics 示例metrics.py系列建议的进阶路线先在concurrent_execution.py与retries.py上跑通基础鲁棒性再通过basic_agent_events.py建立可观测性随后按业务需求引入advanced_compression.py长会话与background_execution.py耗时任务最后用agent_serialization.py固化 Agent 配置。所有示例均可在完成 cookbook 环境准备 后用.venvs/demo/bin/python cookbook/02_agents/14_advanced/file.py直接运行验证。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/10 6:19:02 · 阅读 1.2k Agno Agent 高级实战指南：缓存、压缩、并发、事件、重试、调试与序列化
Agno Agent 高级实战指南缓存、压缩、并发、事件、重试、调试与序列化【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇指南聚焦 Agnoagno仓库中 cookbook/02_agents/14_advanced 这一高级专题目录系统讲解 Agent 在生产级场景下的七大进阶能力模型响应缓存、上下文压缩、并发与后台执行、运行取消与持久化、生命周期事件监听、重试与调试日志、以及 Agent 序列化。读完本文你将能够基于仓库中的可运行示例直接为自己的 Agent 应用叠加这些工程能力并理解其底层实现位置。目录概览与文件清单14_advanced目录围绕生产可用展开覆盖缓存、压缩、并发、事件、重试、调试与序列化七大主题。以下是目录内全部示例文件的职责说明文件主题核心能力cache_model_response.py缓存缓存模型响应避免重复 API 调用advanced_compression.py压缩基于 token 上限的工具调用上下文压缩tool_call_compression.py压缩压缩工具调用结果控制上下文体积compression_events.py压缩/事件压缩过程中的事件流监听concurrent_execution.py并发用asyncio.gather并发执行多个任务background_execution.py后台后台运行 轮询 取消background_execution_structured.py后台带结构化输出的后台执行background_execution_concurrency.py后台/并发后台并发执行background_execution_metrics.py后台/指标后台运行的指标采集background_streaming_resume.py后台/流式后台流式执行与断点恢复redis_event_stream_resume.py事件/恢复基于 Redis 事件流的恢复cancel_run.py取消取消正在运行的 Agentcustom_cancellation_manager.py取消自定义取消逻辑agent_run_cancel_persistence.py取消/持久化取消运行并验证部分内容落库basic_agent_events.py事件监听 Agent 生命周期事件reasoning_agent_events.py事件推理步骤中的事件retries.py重试带指数退避的重试配置debug.py调试开启 verbose 调试输出custom_logging.py日志自定义日志配置metrics.py/multi_model_metrics.py/session_metrics.py/session_summary_metrics.py/streaming_metrics.py/tool_call_metrics.py指标访问 Agent 运行指标agent_serialization.py序列化Agent 的序列化与反序列化interchange_model/模型互操作跨提供商的模型互换OpenAI/Claude/Gemini 等运行环境与前置条件在运行本目录任何示例之前需要完成以下准备加载环境变量在仓库根目录执行direnv allow确保OPENAI_API_KEY等密钥可用。创建演示环境运行./scripts/demo_setup.sh创建虚拟环境之后所有 cookbook 均通过.venvs/demo/bin/python执行。可选依赖部分示例依赖本地服务或特定提供商的 API 密钥。例如agent_run_cancel_persistence.py与background_execution.py要求本地 PostgreSQLpostgresqlpsycopg://ai:ailocalhost:5532/ai可参照仓库内cookbook/scripts/run_pgvector.sh启动。运行单个示例的统一命令格式为.venvs/demo/bin/python cookbook/02_agents/14_advanced/file.py例如.venvs/demo/bin/python cookbook/02_agents/14_advanced/cache_model_response.py模型响应缓存消除重复 API 调用cache_model_response.py演示了 Agno 最直接的优化手段对相同请求复用模型响应从而减少 API 调用与延迟。from agno.agent import Agent from agno.models.openai import OpenAIResponses # 在模型层开启响应缓存 agent Agent(modelOpenAIResponses(idgpt-5.6-luna, cache_responseTrue)) # 同一查询执行两次第一次为 Cache Miss第二次为 Cache Hit for i in range(1, 3): response agent.run( Write me a short story about a cat that can talk and solve problems. ) print(response.content) print(f\n Elapsed time: {response.metrics.duration:.3f}s)关键点缓存开关位于模型层通过OpenAIResponses(..., cache_responseTrue)开启而不是在 Agent 上配置。这是理解缓存粒度的关键——缓存与具体模型实例绑定。验证方式示例连续两次运行相同 query并在输出中标注 Cache Miss (First Request) 与 Cache Hit (Cached Response)同时通过response.metrics.duration对比两次耗时。缓存命中时耗时通常显著下降。注意缓存语义依赖具体模型提供商的实现与密钥配置示例中的gpt-5.6-luna为仓库当前示例所用模型 ID实际使用时应替换为你环境内可用的模型。上下文压缩控制长会话与工具调用的 token 消耗上下文窗口是 Agent 应用的硬约束。14_advanced提供了三级压缩方案核心入口是 libs/agno/agno/compression 模块中的CompressionManager。基于 token 上限的工具调用压缩advanced_compression.py该示例为竞品情报分析师场景定制了压缩策略当上下文达到指定 token 阈值时对工具调用历史进行压缩。from agno.agent import Agent from agno.compression.manager import CompressionManager from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.tools.websearch import WebSearchTools compression_prompt You are a compression expert. Your goal is to compress web search results for a competitive intelligence analyst. MUST PRESERVE: - Competitor names and specific actions (product launches, partnerships, acquisitions, pricing changes) - Exact numbers (revenue, market share, growth rates, pricing, headcount) - Precise dates (announcement dates, launch dates, deal dates) - Direct quotes from executives or official statements - Funding rounds and valuations MUST REMOVE: - Company history and background information - General industry trends (unless competitor-specific) - Analyst opinions and speculation (keep only facts) ... OUTPUT FORMAT: Return a bullet-point list where each line follows this format: [Company Name] - [Date]: [Action/Event] ([Key Numbers/Details]) Keep it under 200 words total. Be ruthlessly concise. Facts only. compression_manager CompressionManager( modelOpenAIResponses(idgpt-5-mini), compress_token_limit5000, # 上下文 token 超过该阈值即触发压缩 compress_tool_call_instructionscompression_prompt, # 领域定制的压缩指令 ) agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[WebSearchTools()], descriptionSpecialized in tracking competitor activities, instructionsUse the search tools and always use the latest information and data., dbSqliteDb(db_filetmp/token_based_tool_call_compression.db), compression_managercompression_manager, add_history_to_contextTrue, # 将历史加入上下文 num_history_runs3, # 保留最近 3 轮历史 session_idtoken_based_tool_call_compression, )参数要点compress_token_limit触发压缩的上下文 token 阈值此处为 5000。这是基于 token 的压缩的核心区别于按轮次/数量压缩。compress_tool_call_instructions领域定制提示词示例通过MUST PRESERVE保留竞品名称、精确数字、日期、高管原话、融资轮次与MUST REMOVE剔除公司历史、行业泛论、分析师观点、营销话术来确保压缩后信息无损于业务目标。这展示了压缩指令模板化的最佳实践。配套存储示例使用SqliteDb持久化会话配合add_history_to_contextTrue与num_history_runs3使压缩与多轮历史协同工作。工具调用结果压缩tool_call_compression.pytool_call_compression.py聚焦单次工具调用结果体积过大的问题——例如 Web 搜索返回大量页面摘要时可在写入上下文前对工具结果进行压缩。它与advanced_compression.py的区别在于粒度前者压缩历史的工具调用后者针对即将进入上下文的工具结果。压缩过程事件compression_events.pycompression_events.py演示在压缩发生时订阅事件流用于观察压缩的触发时机、被压缩的内容量以及压缩后的 token 节省情况。这在调优compress_token_limit时非常有用——你可以据此判断阈值是否过松频繁压缩或过紧上下文仍超限。从源码结构看压缩逻辑统一由 libs/agno/agno/compression/manager.py 中的CompressionManager调度其核心职责是监测上下文 token 用量 → 达到阈值时调用压缩模型 → 用压缩结果替换原始工具调用历史 → 触发压缩事件。compress_token_limit与压缩指令均作为该管理器构造参数传入。并发执行单个 Agent 实例并行处理多任务concurrent_execution.py演示了通过asyncio.gather让同一个 Agent 实例并发执行多个独立研究任务import asyncio from agno.agent import Agent from agno.models.openai import OpenAIResponses from agno.tools.duckduckgo import DuckDuckGoTools providers [openai, anthropic, ollama, cohere, google] instructions Your task is to write a well researched report on AI providers. The report should be unbiased and factual. # 在循环外只创建一次 Agent —— 这是正确的并发模式 agent Agent( modelOpenAIResponses(idgpt-5-mini), instructionsinstructions, tools[DuckDuckGoTools()], ) async def get_reports(): 使用同一个 Agent 实例并发运行多个研究任务。 tasks [ agent.arun(fWrite a report on the following AI provider: {provider}) for provider in providers ] results await asyncio.gather(*tasks) return results async def main(): results await get_reports() for result in results: pprint(result.content) if __name__ __main__: asyncio.run(main())实践要点复用实例而非复制实例示例注释明确指出Create the agent ONCE outside the loop - this is the correct pattern。并发安全的前提是共享同一个Agent对象调用其异步接口arun。异步 APIagent.arun(...)返回协程由asyncio.gather统一调度5 个研究任务并行推进。适用边界并发任务之间应相互独立无共享可变状态且底层模型 API 需支持并发请求配额。后台执行提交即返回轮询或取消background_execution.py展示了后台执行的完整生命周期提交backgroundTrue的 run 后立即返回PENDING状态实际工作在后台推进调用方随后轮询结果或随时取消。示例一后台运行 轮询db PostgresDb( db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai, session_tablebackground_exec_sessions, ) agent Agent( nameBackgroundAgent, modelOpenAIResponses(idgpt-5-mini), descriptionAn agent that runs in the background, dbdb, ) # 后台启动 —— 立即返回 PENDING run_output await agent.arun( What is the capital of France? Answer in one sentence., backgroundTrue, ) assert run_output.status RunStatus.pending # 每秒轮询一次直至 completed / error for i in range(30): await asyncio.sleep(1) result await agent.aget_run_output( run_idrun_output.run_id, session_idrun_output.session_id, ) if result.status RunStatus.completed: print(f\nCompleted! Content: {result.content}) break elif result.status RunStatus.error: break示例二与示例三取消后台运行与先取消后启动# 取消一个正在运行的后台任务 cancelled await agent.acancel_run(run_idrun_output.run_id) # 取消-先于-启动语义预生成 run_id在启动前取消 from agno.run.cancel import cancel_run from uuid import uuid4 run_id str(uuid4()) cancel_run(run_id) # 先取消 run_output await agent.arun( This should be cancelled before it runs., backgroundTrue, run_idrun_id, # 启动时检测到已取消 )关键机制状态机后台运行的状态由agno.run.base.RunStatus定义pending→completed/error/cancelled。提交后立即返回pending随后在数据库中更新。轮询接口aget_run_output(run_id, session_id)按 run_id session_id 从数据库读取最新状态。取消语义acancel_run取消进行中的运行agno.run.cancel.cancel_run(run_id)支持先取消后启动——即便 run 尚未开始也可以在启动时被检测到并取消。这是分布式/长时间运行场景下的重要容错能力。存储前提后台执行依赖数据库示例用 PostgreSQL记录 run 状态无存储时该能力不可用。从源码结构看后台执行的调度与状态查询实现在 libs/agno/agno/run 目录下background_execution.py是backgroundTrue参数与运行状态存储之间的完整调用链示例。目录中还有background_execution_structured.py后台 Pydantic 结构化输出、background_execution_concurrency.py后台并发、background_execution_metrics.py后台指标、background_streaming_resume.py后台流式 恢复以及redis_event_stream_resume.py基于 Redis 事件流恢复运行可按需组合上述模式。运行取消中断长任务与部分内容持久化基础取消cancel_run.pycancel_run.py通过双线程演示取消流程一个线程运行长任务写 2000 字以上的长故事并流式消费内容另一个线程在延迟后调用agent.cancel_run(run_id)for chunk in agent.run(...长故事提示词..., streamTrue): if run_id not in run_id_container and chunk.run_id: run_id_container[run_id] chunk.run_id if chunk.event RunEvent.run_content: content_pieces.append(chunk.content) print(chunk.content, end, flushTrue) # 运行被取消时会发出 RunEvent.run_cancelled 事件 elif chunk.event RunEvent.run_cancelled: print(f\n[CANCELLED] Run was cancelled: {chunk.run_id}) return elif hasattr(chunk, status) and chunk.status RunStatus.completed: final_response chunk # 取消线程 success agent.cancel_run(run_id)要点取消是标记式的cancel_run(run_id)返回布尔值表示取消是否被受理对于已结束或不存在的 run返回失败并给出提示。事件驱动感知取消发生后流中会出现RunEvent.run_cancelled事件业务代码据此收尾示例中保存已生成的部分内容。run_id 传递多线程场景下需通过共享容器示例为run_id_containerdict在运行线程与取消线程间传递run_id。取消 持久化验证agent_run_cancel_persistence.py该示例将取消能力与数据库持久化结合流式运行中收集到 20 个内容块后主动取消随后从会话中取回最后一次 run验证部分内容与消息是否已写入数据库agent Agent( nameStoryteller, modelOpenAIResponses(idgpt-5.4), instructionsYou are a storyteller. Write very long detailed stories., dbPostgresDb(db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai), store_tool_messagesTrue, store_history_messagesTrue, ) for event in agent.run(input..., streamTrue, stream_eventsTrue): if len(content_chunks) 20 and run_id and not cancelled: agent.cancel_run(run_id) cancelled True if event.event RunEvent.run_cancelled: print(\nRun was cancelled) break # 验证从会话中读取最后一条 run 的部分内容与消息 session agent.get_session(session_idagent.session_id) 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 [])})这证明了 Agno 的一个重要设计取消不等于丢失。通过store_tool_messages与store_history_messages开启消息存储后中断的 run 的已生成内容、状态与消息都会被持久化可支撑后续断点续跑或人工审查。custom_cancellation_manager.py则进一步允许替换取消逻辑的实现自定义CancellationManager用于接入自定义的分布式取消信号如 Redis 发布订阅等场景。事件监听洞察 Agent 生命周期basic_agent_events.py展示了如何通过stream_eventsTrue订阅 Agent 的完整生命周期事件这是构建可观测性、日志与 UI 流式展示的基础from agno.agent import RunEvent from agno.agent.agent import Agent finance_agent Agent( idfinance-agent, nameFinance Agent, modelOpenAIResponses(idgpt-5.2), tools[YFinanceTools()], ) async def run_agent_with_events(prompt: str): content_started False async for run_output_event in finance_agent.arun( prompt, streamTrue, stream_eventsTrue ): if run_output_event.event in [RunEvent.run_started, RunEvent.run_completed]: print(f\nEVENT: {run_output_event.event}) if run_output_event.event in [RunEvent.tool_call_started]: print(fTOOL CALL: {run_output_event.tool.tool_name}) print(fTOOL CALL ARGS: {run_output_event.tool.tool_args}) if run_output_event.event in [RunEvent.tool_call_completed]: print(fTOOL CALL: {run_output_event.tool.tool_name}) print(fTOOL CALL RESULT: {run_output_event.tool.result}) if run_output_event.event in [RunEvent.run_content]: print(run_output_event.content, end)事件体系要点事件枚举核心生命周期事件定义在agno.run.agent.RunEvent本示例从agno.agent导入RunEvent覆盖run_started、run_completed、tool_call_started、tool_call_completed、run_content、run_cancelled等。取消示例中使用的RunEvent.run_cancelled与压缩示例中的压缩事件同属该体系。事件负载工具调用事件携带tool对象tool_name、tool_args、result内容事件携带content流式片段每个事件还携带run_id取消/持久化示例正是利用这一点拿到 run_id。推理事件reasoning_agent_events.py补充了推理步骤reasoning steps期间的事件适用于需要展示模型思考过程的可解释性场景。重试指数退避提升鲁棒性retries.py演示了网络抖动、限流等临时错误下的自动重试agent Agent( nameWeb Search Agent, roleSearch the web for information, tools[WebSearchTools()], retries3, # 出错时整个 Agent run 最多重试 3 次 delay_between_retries1, # 每次重试之间的基础延迟秒 exponential_backoffTrue, # 开启后延迟每次翻倍 )参数语义retriesAgent run 在出错情况下的最大重试次数示例为 3。delay_between_retries基础延迟秒数。exponential_backoff为 True 时每次重试的延迟翻倍1s → 2s → 4s这是对 API 限流类错误最友好的退避策略避免雪崩式重试。调试与自定义日志开启调试模式debug.py# 方式一Agent 级全局开启所有 run 输出更详细的日志 agent Agent(modelOpenAIResponses(idgpt-5-mini), debug_modeTrue) agent.print_response(inputTell me a joke.) # 方式二单次 run 临时开启 agent Agent(modelOpenAIResponses(idgpt-5-mini)) agent.print_response(inputTell me a joke., debug_modeTrue)debug_modeTrue会产生更 verbose 的输出请求/响应细节、内部调用链适合排查 Agent 行为异常。注意单次 run 的debug_mode参数优先级高于 Agent 级配置可在不污染全局配置的前提下做一次性的深度排查。自定义日志custom_logging.pyAgno 的日志统一由 libs/agno/agno/utils/log.py 提供可通过configure_agno_logging无缝替换为自有 loggerimport logging from agno.utils.log import configure_agno_logging, log_info def get_custom_logger(): custom_logger logging.getLogger(custom_logger) handler logging.StreamHandler() formatter logging.Formatter([CUSTOM_LOGGER] %(levelname)s: %(message)s) handler.setFormatter(formatter) custom_logger.addHandler(handler) custom_logger.setLevel(logging.INFO) custom_logger.propagate False return custom_logger custom_logger get_custom_logger() # 全局替换此后 Agno 的所有日志均走自定义 logger configure_agno_logging(custom_default_loggercustom_logger) log_info(This is using our custom logger!) agent Agent() agent.print_response(What can I do to improve my sleep?)要点configure_agno_logging(custom_default_logger...)是一次性全局配置之后agno.utils.log中所有日志函数如log_info以及 Agent 内部日志都会路由到自定义 logger。这是将 Agno 接入企业日志体系JSON 结构化、日志采集、级别策略的标准入口。Agent 序列化保存、加载与重建agent_serialization.py演示了 Agent 的三种物化方式字典互转、按版本持久化到数据库、从数据库加载from agno.agent import Agent from agno.db.sqlite import SqliteDb agent_db SqliteDb(db_filetmp/agents.db) agent Agent( idserialization-demo-agent, nameSerialization Demo Agent, modelOpenAIResponses(idgpt-5.2), dbagent_db, ) # 方式一与 dict 互转内存内重建 config agent.to_dict() recreated Agent.from_dict(config) # 方式二按版本保存到数据库再按 id version 加载 version agent.save() loaded Agent.load(idagent.id, dbagent_db, versionversion) recreated.print_response(Say hello from a recreated agent., streamTrue) loaded.print_response(Say hello from a loaded agent., streamTrue)三种途径的适用场景to_dict()/Agent.from_dict(config)轻量重建适合配置迁移、复制 Agent 模板或在不依赖数据库的环境间传递 Agent 定义。agent.save()→version将 Agent 序列化后写入数据库此处为SqliteDb数据库无关并返回版本号。Agent.load(id..., db..., version...)按唯一 id 与版本号还原历史 Agent 配置适合版本管理与审计。序列化能力由 libs/agno/agno/agent 中的 Agent 定义支撑结合任意 Agno 支持的数据库SQLite/PostgreSQL 等即可获得配置即数据的持久化体验。运行指标量化 Agent 行为目录下metrics.py、multi_model_metrics.py、session_metrics.py、session_summary_metrics.py、streaming_metrics.py、tool_call_metrics.py及background_execution_metrics.py构成一套完整的指标观察矩阵分别从以下维度量化运行单次 run 指标metrics.py通过response.metrics.duration等字段获取耗时缓存示例已用到。流式指标streaming_metrics.py衡量首 token 延迟与整体流式传输耗时。工具调用指标tool_call_metrics.py统计各工具调用的次数与耗时定位瓶颈工具。会话与汇总指标session_metrics.py、session_summary_metrics.py跨多次 run 聚合评估长会话的整体成本与表现。多模型对比multi_model_metrics.py在同一任务上横向对比不同模型的耗时与输出服务于选型。后台运行指标background_execution_metrics.py后台任务同样暴露指标便于无人值守场景的监控采集。这些指标对象可直接对接自建监控看板或日志采集管线与上文的事件体系互补——事件描述发生了什么指标量化花了多少成本/时间。小结七大高级能力速查能力关键 API / 参数代表示例模型响应缓存OpenAIResponses(cache_responseTrue)cache_model_response.py上下文压缩CompressionManager(compress_token_limit, compress_tool_call_instructions)advanced_compression.py并发执行agent.arun(...)asyncio.gatherconcurrent_execution.py后台执行arun(backgroundTrue)aget_run_output轮询 acancel_runbackground_execution.py运行取消agent.cancel_run(run_id)RunEvent.run_cancelledcancel_run.py、agent_run_cancel_persistence.py事件监听stream_eventsTrueRunEvent.*basic_agent_events.py重试retries、delay_between_retries、exponential_backoffretries.py调试/日志debug_modeTrue、configure_agno_logging(custom_default_logger...)debug.py、custom_logging.py序列化to_dict()/from_dict()、save()/load()agent_serialization.py指标response.metrics、各专项 metrics 示例metrics.py系列建议的进阶路线先在concurrent_execution.py与retries.py上跑通基础鲁棒性再通过basic_agent_events.py建立可观测性随后按业务需求引入advanced_compression.py长会话与background_execution.py耗时任务最后用agent_serialization.py固化 Agent 配置。所有示例均可在完成 cookbook 环境准备 后用.venvs/demo/bin/python cookbook/02_agents/14_advanced/file.py直接运行验证。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考