Agno 工作流中的历史History管理从对话连续执行到跨轮次上下文感知【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno 的Workflow除了支持多步骤编排之外还提供了一套完整的工作流历史机制让每一步执行都能感知同一会话session中此前的多轮运行结果。本篇基于 cookbook/04_workflows/06_advanced_concepts/history 下的 4 个可运行示例讲解如何在单步对话、自定义函数步骤、意图路由和步骤级粒度上启用并利用工作流历史并结合 Agno 源码说明其底层实现原理。一、什么是工作流历史为什么需要它在普通的单次运行场景里Workflow的每个步骤只看到当前这一次的输入与前面步骤的输出。但很多真实业务AI 家教、内容创作流水线、客服系统、饮食规划助手需要跨轮次记忆用户上一轮说过我昨天吃了意大利菜这一轮要求推荐健康食谱如果工作流无法回看上一轮历史推荐结果就会重复或跑偏。Agno 为此提供两级开关工作流级Workflow(add_workflow_history_to_stepsTrue, ...)对所有步骤统一启用历史注入步骤级Step(..., add_workflow_historyTrue)只对选中的步骤注入历史实现细粒度控制。两条开关都以会话为基础历史数据持久化在数据库如 SQLite中因此即使工作流进程重启只要使用相同的session_id历史依然可回放。源码 workflow.py 中add_workflow_history_to_steps: bool False默认关闭并且源码在 workflow.py 处做了强校验启用历史但未配置db时会直接报错因为历史必须依赖会话存储。运行 4 个示例前需要先激活演示环境并加载 API Key示例使用 OpenAI 模型# 激活 demo 环境 .venvs/demo/bin/python # 加载 API keys需要本地 .envrc 文件 direnv allow二、单步对话式执行continuous_execution.pycontinuous_execution.py 演示的是最简单却最实用的场景一个 AI 家教工作流只有单个步骤但借助历史实现多轮连续对话。核心代码如下tutor_agent Agent( nameAI Tutor, modelOpenAIChat(idgpt-5.6-luna), instructions[ You are an expert tutor who provides personalized educational support., You have access to our full conversation history., Build on previous discussions - do not repeat questions or information., Reference what the student has told you earlier in our conversation., Adapt your teaching style based on what you have learned about the student., When asked about conversation history, provide a helpful summary., Focus on helping the student understand concepts and improve their skills., ], ) tutor_workflow Workflow( nameSimple AI Tutor, descriptionSingle-step conversational tutoring with history awareness, dbSqliteDb(db_filetmp/simple_tutor_workflow.db), steps[Step(nameAI Tutoring, agenttutor_agent)], add_workflow_history_to_stepsTrue, )关键点有三个Agent 提示词面向历史设计instructions明确告诉模型你拥有完整对话历史不要重复提问引用学生此前说过的话历史注入的效果通过提示词放大add_workflow_history_to_stepsTrue工作流把历史塞进步骤输入单步 Agent 也能记住整段会话持久化会话SqliteDb(db_filetmp/simple_tutor_workflow.db)把会话与历史落到本地文件配合固定的session_id实现跨进程续聊。运行入口demo_simple_tutoring_cli()使用workflow.cli_app(...)启动交互式命令行tutor_workflow.cli_app( session_idsimple_tutor_demo, # 固定会话 ID历史按此聚合 userStudent, emoji, streamTrue, show_step_detailsTrue, # 展示每个步骤的详细信息 )实际体验路径是学生先问我在导数上遇到困难AI 给出讲解下一轮学生说能帮我学代数吗Agent 会引用前一轮背景、按已知的学生画像调整教学方式——这正是单步工作流持续执行的含义工作流本身每轮只跑一步但上下文在轮次之间流动。三、在自定义函数步骤中读取历史history_in_function.pyAgent 步骤依赖 LLM 消化历史而 history_in_function.py 展示了更硬核的用法在纯 Python 函数步骤executor中用代码结构化地读取工作流历史并据此做出确定性决策。该示例构建了一个研究 → 策略分析 → 写作的三步内容创作工作流其中间步骤是自定义函数def analyze_content_strategy(step_input: StepInput) - StepOutput: current_topic step_input.input or research_data step_input.get_last_step_content() or # 上一步研究的输出 history_data step_input.get_workflow_history(num_runs5) # 最近 5 轮的历史 ...这里用到StepInput上两个关键 API源码定义见 types.pyget_last_step_content()返回本步骤之前最近一个步骤的输出内容保持向后兼容的便捷方法在本例中即Content Research步骤产出的调研文本get_workflow_history(num_runs5)以List[Tuple[str, str]]形式返回最近 5 轮运行的(输入, 输出)对供函数做结构化分析。函数内部实现了一套轻量策略引擎对当前主题做关键词抽取含同义词映射如 ai→artificial/intelligence再与历史各轮的输入做关键词重叠度计算输出重叠百分比、内容多样性得分与策略建议overlap_percentage (topic_overlap / max(max_possible_overlap, 1)) * 100 diversity_score len(set(covered_topics)) / max(len(covered_topics), 1) recommendations [] if overlap_percentage 60: recommendations.append(HIGH OVERLAP detected - consider a fresh angle or advanced perspective) elif overlap_percentage 30: recommendations.append(MODERATE OVERLAP detected - differentiate your approach) if diversity_score 0.6: recommendations.append(Low content diversity - explore different aspects of the topic)最终函数把分析结果打包成结构化StepOutput含历史覆盖情况、避重建议、建议切入角度等交给下游Strategic Content CreationAgent 步骤使用return StepOutput(contentformatted_analysis.strip())工作流的组装方式return Workflow( nameStrategic Content Creation, descriptionResearch - Strategic Analysis - Content Creation with historical awareness, dbSqliteDb(db_filetmp/content_workflow.db), steps[research_step, strategy_step, writer_step], add_workflow_history_to_stepsTrue, )这个例子的价值在于当业务需要确定性判断而不是让 LLM 自由发挥时可以用函数步骤把历史变成可计算的指标——例如防止内容团队重复生产同一主题、衡量选题多样性这在内容运营与选题规划类产品中可直接落地。四、意图路由 共享历史intent_routing_with_history.pyintent_routing_with_history.py 把历史机制与路由结合构建了一个智能客服工作流同一个用户会话在多轮之间可能被路由到不同专员而每个专员都要看到完整对话历史。三个专员 Agent 分别是tech_support_agent技术支持、billing_agent账单与账户、general_support_agent通用客服每个 Agent 的提示词都强调你有完整对话历史、要引用之前的交互、要承接已尝试的排查步骤。对应的步骤显式开启历史tech_support_step Step( nameTechnical Support, agenttech_support_agent, add_workflow_historyTrue, # 步骤级开启历史 ) billing_support_step Step(nameBilling Support, agentbilling_agent, add_workflow_historyTrue) general_support_step Step(nameGeneral Support, agentgeneral_support_agent, add_workflow_historyTrue)路由逻辑是纯函数的simple_intent_router基于关键词做意图判定def simple_intent_router(step_input: StepInput) - List[Step]: current_message step_input.input or current_message_lower current_message.lower() tech_keywords [api, error, bug, technical, login, not working, broken, crash] billing_keywords [billing, payment, refund, charge, subscription, invoice, plan] if any(keyword in current_message_lower for keyword in tech_keywords): print(Routing to Technical Support) return [tech_support_step] if any(keyword in current_message_lower for keyword in billing_keywords): print(Routing to Billing Support) return [billing_support_step] print(Routing to General Support) return [general_support_step]工作流通过Router步骤把路由器和候选步骤组装起来同时在工作流级开启历史兜底return Workflow( nameSmart Customer Service, descriptionSimple routing to specialists with shared conversation history, dbSqliteDb(db_filetmp/smart_customer_service.db), steps[ Router( nameCustomer Service Router, selectorsimple_intent_router, choices[tech_support_step, billing_support_step, general_support_step], descriptionRoutes to appropriate specialist based on simple intent detection, ) ], add_workflow_history_to_stepsTrue, )从源码看步骤级与工作流级开关的优先级是Step.add_workflow_history若显式设置则优先否则回退到add_workflow_history_to_steps见 step.py。因此这里的语义是三个专员步骤各自显式开启历史工作流级开关作为默认值兜底。典型对话流用户先问我的 API 不工作了→ 路由到技术支持用户下一轮说顺便问下退款政策→ 路由到账单专员此时账单专员也能看到上一轮技术支持已排查到哪一步实现换人不断线的客服体验。五、工作流级与步骤级历史控制step_history.pystep_history.py 是本目录信息量最大的示例它同时演示了工作流级历史与步骤级历史两种模式的对比并引入了一个自定义偏好分析函数。5.1 工作流级历史会话式饮食规划Conversational Meal Planner工作流由三个步骤组成suggestion_step餐品建议、preference_analysis_step偏好分析函数步骤、recipe_step食谱推荐。它在工作流级统一开启历史meal_workflow Workflow( nameConversational Meal Planner, descriptionSmart meal planning with conversation awareness and preference learning, dbSqliteDb(session_tableworkflow_session, db_filetmp/meal_workflow.db), steps[suggestion_step, preference_analysis_step, recipe_step], add_workflow_history_to_stepsTrue, )其中的偏好分析函数用step_input.previous_step_content即get_last_step_content()的别名属性拿到上一步建议内容并结合当前请求做关键词规则解析def analyze_food_preferences(step_input: StepInput) - StepOutput: current_request step_input.input conversation_context step_input.previous_step_content or ... if italian in full_context and (had in full_context or yesterday in full_context): preferences[avoid_list].append(Italian) ... return StepOutput(contentanalysis_result)配合demonstrate_conversational_meal_planning()中同一个session_id下的连续三次print_response调用可以看到历史机制如何逐步积累第一轮今晚吃什么→ 第二轮我昨天吃了意大利菜最近想吃得健康些函数步骤从previous_step_content与历史中识别出 avoid 列表→ 第三轮想吃鱼我也喜欢亚洲风味偏好继续累加。三轮共享meal_planning_demo这个会话 ID是历史跨轮次生效的关键前提。5.2 步骤级历史内容创作流水线Smart Content Creation Pipeline工作流刻意演示了选择性历史注入只有 Research 和 Content Creation 两个步骤开启历史最后的 Publishing 步骤不感知历史content_workflow Workflow( nameSmart Content Creation Pipeline, descriptionResearch - Content Creation (with history awareness) - Publishing, dbSqliteDb(db_filetmp/content_workflow.db), steps[ Step(nameResearch Phase, agentresearch_agent, add_workflow_historyTrue), Step(nameContent Creation, agentcontent_creator, add_workflow_historyTrue), Step(nameContent Publishing, agentpublisher_agent), ], )这一步级控制非常实用content_creator的提示词要求Review workflow history to understand what content topics have been covered before, what writing styles were used previously, avoid repeating similar content即只有真正需要避免重复创作的步骤才消耗历史上下文而发布步骤只需处理当轮产物注入历史反而浪费 token 并可能干扰输出格式。与 5.1 对比可得到清晰的选型建议全局一致需要记忆的用工作流级开关只有个别步骤需要记忆的用步骤级开关。六、源码级原理历史如何被注入与读取从源码可以勾勒出工作流历史机制的完整链路配置校验工作流构造时若add_workflow_history_to_stepsTrue但未设置dbworkflow.py 抛出异常步骤级同理workflow.py 会在运行前校验step.add_workflow_historyTrue的步骤所在工作流必须有数据库。历史能力与会话持久化强绑定。历史上下文注入步骤运行前若步骤自身的add_workflow_history为真或未设置时回退到工作流级开关会调用workflow_session.get_workflow_history_context(num_runsnum_history_runs)取历史并拼入步骤输入见 step.pyAgent 步骤由此获得完整对话历史。结构化读取 API对于函数步骤StepInput.get_workflow_history(num_runsNone)返回List[Tuple[str, str]]每轮的用户输入与工作流输出get_workflow_history_context()返回格式化文本get_last_step_content()返回最近一个前置步骤的内容见 types.py。两者一个面向代码计算一个面向拼进提示词覆盖两类消费方式。会话维度历史按session_id聚合这正是所有示例都使用固定session_id如content_strategy_demo的原因。数据库选用SqliteDb(db_filetmp/*.db)换用 Postgres、MySQL 等存储只需替换db参数历史读写逻辑不变。七、总结与适用场景工作流历史让 Agno 的Workflow从单次执行的流水线升级为跨轮次有记忆的对话式编排。回顾 4 个示例的适用边界示例核心能力适用场景continuous_execution.py单步 Agent 工作流级历史AI 家教、顾问式对话、需要长期记忆的单 Agent 服务history_in_function.py函数步骤结构化读取历史内容去重、选题策略、需要确定性计算的历史分析intent_routing_with_history.py路由 步骤级历史共享客服分流、多技能组协作、跨专员上下文衔接step_history.py工作流级 vs 步骤级对比需要精细控制哪些步骤该看到历史的流水线一个可复用的工程决策若全部步骤都应感知历史直接Workflow(add_workflow_history_to_stepsTrue)若仅个别步骤需要用Step(add_workflow_historyTrue)精打细算上下文窗口若需要程序化判断历史内容则在函数步骤中使用get_workflow_history()与get_last_step_content()。无论哪种方式都要先配置db并保持session_id稳定历史才能跨轮次、跨进程持续生效。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考