agno Workflow 会话状态Session State实战用可持久化状态构建多轮、跨步骤的自适应工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本指南围绕 agno 仓库中 cookbook/04_workflows/06_advanced_concepts/session_state 的 7 个可运行示例展开系统讲解 Workflow 的会话状态session_state机制如何在普通 Agent 步骤、Team 步骤、自定义 executor、条件判断Condition与路由Router之间读写同一份状态并借助 SqliteDb 等存储将状态跨运行持久化。读完你既能掌握与 agent 工具调用、结构化输出结合的累积型状态应用如职位申请跟踪器、任务清单也能看懂会话命名、新老用户问候、偏好自适应路由等典型多轮交互场景的落地写法。什么是 Workflow 会话状态Workflow 在运行一条input时会产生一次 run而多个 run 又归属于同一个 session。session_state就是这个 session 级别的、以 Python 字典形式存在的共享工作区它可以被 workflow 中任意步骤的 executorAgent、Team、自定义函数读取或改写并随 session 一起存储从而跨越步骤边界与运行边界保留数据。在源码层面Workflow 类的定义libs/agno/agno/workflow/workflow.py中session_state是被显式声明与构造参数化的字段# Default session state (stored in the database to persist across runs) session_state: Optional[Dict[str, Any]] None # Set to True to overwrite the stored session_state with the session_state provided in the run overwrite_db_session_state: bool False在初始化时传入的session_state{applications: []}这类写法相当于为会话状态提供了初始 schema / 默认键值。默认情况下它只作为首次的种子值若数据库中已存在该 session 的历史状态运行时会优先加载历史值overwrite_db_session_stateFalse保证多轮对话间的数据连续性只有在显式打开该开关时才会用本次传入的字典覆盖库中已存状态。工具函数、executor 等通过RunContext各 executor 的run_context参数访问这份字典run_context.session_state # 一个可原地读写的 dict运行结束后任何一步写入的键值都会保存在 session 中可通过workflow.get_session_state()读取从而在脚本中验证、打印或二次消费状态。环境与运行前提本目录示例的运行方式在 README.md 中给出使用 demo 虚拟环境执行.venvs/demo/bin/python通过direnv allow加载 API Key需要本地存在.envrc文件示例默认使用 OpenAI 系模型OpenAIChat/OpenAIResponses需保证网络与密钥可用。其中job_application_tracker.py的执行记录见 TEST_LOG.md验证了它的跨进程持久化行为同一固定 session 下跟踪器中的记录数随三次运行由 1 → 2 → 3 递增且在新进程重新读取时仍能取回相同记录确认了 SqliteDb 的持久化。其余示例为一次性演示脚本部分在测试环境中因单次 35 秒超时而中断属超时限制而非逻辑错误本地运行可放宽时间预期。第一个例子Agent 步骤间共享状态state_with_agent.py最基础的能力是同一个 session 内多个 Agent 步骤共享同一份可变字典。示例 state_with_agent.py 实现了一个购物清单工作流。核心做法是把读写状态封装成普通 Python 函数并以RunContext作为第一个形参注册为 Agent 的工具from agno.run import RunContext def add_item(run_context: RunContext, item: str) - str: if run_context.session_state is None: run_context.session_state {} existing_items [ existing_item.lower() for existing_item in run_context.session_state[shopping_list] ] if item.lower() not in existing_items: run_context.session_state[shopping_list].append(item) return fAdded {item} to the shopping list. return f{item} is already in the shopping list.remove_item、remove_all_items、list_items同样直接对run_context.session_state[shopping_list]做增删查。随后这些函数以tools[...]参数挂到 Agent 上而 Agent 又通过Step挂进 Workflowshopping_assistant Agent( nameShopping Assistant, modelOpenAIChat(idgpt-5.2), tools[add_item, remove_item, list_items], instructions[ You are a helpful shopping assistant., You can help users manage their shopping list by adding, removing, and listing items., Always use the provided tools to interact with the shopping list., ], ) manage_items_step Step( namemanage_items, descriptionHelp manage shopping list items (add/remove), agentshopping_assistant, ) shopping_workflow Workflow( nameShopping List Workflow, dbSqliteDb(db_filetmp/workflow.db), steps[manage_items_step, view_list_step], session_state{shopping_list: []}, # 初始状态种子 )运行后脚本通过shopping_workflow.get_session_state()打印当前状态。示例依次执行加 milk/bread/eggs、加 apples/bananas 并列出、移除 bread、清空清单四轮对话每一轮的变更都沉淀在同一个shopping_list里。这就是Agent 自主决定何时通过工具写状态的典型形态——状态读写完全由模型驱动的工具调用来编排。结合结构化输出与工具调用职位申请跟踪器job_application_tracker.pystate_with_agent.py 演示的是纯工具读写job_application_tracker.py 则更进一步把结构化输出抽取output_schema与会话状态工具串联成流水线并演示了固定session_id让记录跨多次独立运行持续累积。首先定义结构化模型Pydanticclass JobApplication(BaseModel): Structured representation of a single job application. company: str Field(..., descriptionCompany name) role: str Field(..., descriptionJob title / role) url: Optional[str] Field(None, descriptionJob posting URL) source: Optional[str] Field(None, descriptionWhere the job was found, e.g. a job board) status: str Field( defaultApplied, descriptionfApplication status, one of: {, .join(VALID_STATUSES)}, ) notes: Optional[str] Field(None, descriptionAny extra notes)两个工具save_application/list_applications维护run_context.session_state中的applications数组其中setdefault的写法保证了即使session_state尚未初始化也能安全写入def save_application( run_context: RunContext, company: str, role: str, status: str Applied, url: str , source: str , notes: str , ) - str: if run_context.session_state is None: run_context.session_state {} applications run_context.session_state.setdefault(applications, []) record { id: len(applications) 1, company: company, role: role, status: status, url: url, source: source, notes: notes, applied_at: datetime.now().strftime(%Y-%m-%d), } applications.append(record) return fSaved application #{record[id]}: {role} at {company} ({status}).两个 Agent 各司其职extractor_agent用output_schemaJobApplication从用户消息里抽取结构化职位申请tracker_agent挂载上述工具负责保存一次并列出全表。它们分别包装成两个顺序Step进入同一个 Workflowextract_application_step Step( nameextract_application, descriptionExtract a structured job application from the message, agentextractor_agent, ) save_application_step Step( namesave_application, descriptionSave the application to the tracker and list all applications, agenttracker_agent, ) job_application_workflow Workflow( nameJob Application Tracker, dbSqliteDb(db_filetmp/job_application_tracker.db), steps[extract_application_step, save_application_step], session_state{applications: []}, # A fixed session id keeps the tracked applications across separate runs. session_idjob_tracker_demo, )这里最值得注意的是session_idjob_tracker_demo三次print_response调用Backend Engineer、Python Developer、Data Scientist不传session_id会自动回落到 Workflow 上这个固定 id因此每次都命中同一条数据库记录applications数组得以持续累积随后get_session_state()展示出 1 → 2 → 3 条的增长过程。即使重新启动进程、重建 Workflow 对象只要 db 文件与session_id不变历史记录依然可恢复——这是把 Workflow 当作有状态业务系统使用时的标准手法。在自定义 executor 函数中读写状态state_in_function.py有时状态逻辑不需要借助 Agent 工具而是直接写在自定义 Python 函数Step的executor里。示例 state_in_function.py 演示了这种函数式步骤如何读写 session state并额外覆盖了流式 executor返回生成器、逐段产出事件的写法。executor 的签名固定为(step_input: StepInput, run_context: RunContext) - StepOutput。StepInput携带本步输入与上游产物step_input.input是用户原始消息step_input.previous_step_content是前一步输出。状态读写直接落在run_context.session_state上def custom_content_planning_function( step_input: StepInput, run_context: RunContext, ) - StepOutput: session_state run_context.session_state message step_input.input previous_step_content step_input.previous_step_content if content_plans not in session_state: session_state[content_plans] [] if plan_counter not in session_state: session_state[plan_counter] 0 session_state[plan_counter] 1 current_plan_id session_state[plan_counter] # ... 拼接 planning_prompt调用 content_planner.run(...) plan_data { id: current_plan_id, topic: message, content: response.content, has_research: bool(previous_step_content), } session_state[content_plans].append(plan_data) return StepOutput(contentenhanced_content)这种 executor 把编程逻辑 LLM 调用封装成一个步骤。示例的流程是research_stepTeam 或 WebSearch 研究→content_planning_step写入 content_plans→content_summary_step汇总 content_plans 并标记session_summarized/total_plans_summarized。Workflow 同样用session_state{content_plans: [], plan_counter: 0}做种子content_creation_workflow Workflow( nameContent Creation Workflow, dbSqliteDb(session_tableworkflow_session, db_filetmp/workflow.db), steps[research_step, content_planning_step, content_summary_step], session_state{content_plans: [], plan_counter: 0}, )plan_counter自动递增生成 Plan ID跨两次运行AI trends in 2024 与 Machine Learning automation tools持续累计content_plans实现同一 session 内历史计划可追溯、汇总函数可统计的效果。流式版本则把 executor 声明为返回Iterator[Union[WorkflowRunOutputEvent, StepOutput]]的生成器在调用streaming_content_planner.run(..., streamTrue, stream_eventsTrue)时逐事件yield最后再补一个携带状态写入的StepOutput供需要事件级输出的交互场景使用。用状态驱动条件分支新老用户问候state_in_condition.py会话状态不仅能被读写还能反过来决定工作流走向。state_in_condition.py 用状态实现第一次来的用户先被问候老用户直接进入正题。状态判断函数以has_been_greeted为门闩返回布尔值def check_user_has_context(step_input: StepInput, run_context: RunContext) - bool: print(fUser ID: {run_context.session_state.get(current_user_id)}) print(fSession ID: {run_context.session_state.get(current_session_id)}) print(fHas been greeted: {run_context.session_state.get(has_been_greeted, False)}) return run_context.session_state.get(has_been_greeted, False)注意current_user_id、current_session_id可以直接从session_state里取到——框架在注入时会把本次运行的会话标识写入状态上下文从源码结构看运行初始化会构建并注入 RunContextsession_state即为其字段之一。状态写入函数执行后返回StepOutputdef mark_user_as_greeted(step_input: StepInput, run_context: RunContext) - StepOutput: run_context.session_state[has_been_greeted] True run_context.session_state[greeting_count] ( run_context.session_state.get(greeting_count, 0) 1 ) return StepOutput( contentfUser has been greeted. Total greetings: {run_context.session_state[greeting_count]} )组装时把以上两个函数放进Conditionevaluator判定是否为新用户为真时执行Greet Useragent 步骤与Mark as Greetedexecutor 步骤两个子步骤否则跳过之后统一进入Handle Query步骤workflow Workflow( nameConditional Greeting Workflow, steps[ Condition( nameCheck If New User, evaluatorlambda step_input, run_context: ( not check_user_has_context(step_input, run_context) ), steps[ Step(nameGreet User, descriptionGreet the new user, agentgreeter_agent), Step(nameMark as Greeted, descriptionMark user as greeted in session, executormark_user_as_greeted), ], ), Step(nameHandle Query, agentcontextual_agent), ], session_state{has_been_greeted: False, greeting_count: 0}, )两次运行都显式传入session_iduser-123、user_iduser-123。第一次has_been_greetedFalse条件为真、执行问候第二次同一 session 读到True条件短路、跳过问候直接答复。状态因此成了记忆阀门实现了基于会话历史的自适应分支。用状态做自适应路由与任务分拣state_in_router.py路由Router是 Workflow 中选择下一步执行哪个 Step的机制。state_in_router.py 包含两个彼此独立的演示。其一偏好路由。Router 的selector接收(step_input, run_context)返回它选中的Step。选择逻辑同时读取并更新状态def route_based_on_user_preference( step_input: StepInput, run_context: RunContext ) - Step: user_preference run_context.session_state.get(agent_preference, general) interaction_count run_context.session_state.get(interaction_count, 0) run_context.session_state[interaction_count] interaction_count 1 if user_preference technical: return technical_step if user_preference friendly: return friendly_step if interaction_count 0: return onboarding_step return general_step配套的 executorset_user_preference根据interaction_count % 3轮换写入agent_preferencetechnical / friendly / general。二者组合成一个自适应助手adaptive_assistant_workflow Workflow( nameAdaptive Assistant Workflow, steps[ Router( nameRoute to Appropriate Agent, selectorroute_based_on_user_preference, choices[onboarding_step, technical_step, friendly_step, general_step], ), Step(nameUpdate Preferences, executorset_user_preference), ], session_state{agent_preference: general, interaction_count: 0}, )脚本用固定的session_iduser-456依次提问打招呼 → 二叉搜索树 → 披萨话题 → 量子计算首次命中 onboarding之后按偏好与轮换策略在不同 AgentTechnical / Friendly / General之间切换。其二任务分拣路由。这里工具函数先以RunContext操作状态如add_task、complete_task、set_task_priority、list_tasks、clear_completed_tasks共同维护session_state[task_list]含按优先级high/medium/low排序展示、去重、完成态清理等细节路由函数task_router(step_input) - List[Step]则按关键词把请求分发给 Task Manager / Task Viewer / Task Organizertask_workflow Workflow( nameSmart Task Management Workflow, steps[ Router( nametask_management_router, selectortask_router, choices[manage_tasks_step, view_tasks_step, organize_tasks_step], ) ], session_state{task_list: []}, dbSqliteDb(db_filetmp/workflow.db), )五轮对话加任务、查看、完成任务、清理、筛选 pending全部落在同一个task_list上。两个演示合起来说明了 Router 的两种状态接入姿势用状态决定路由目标selector 读状态以及让路由后的工具更新状态executor / 工具写状态。Agent 与 Team 步骤共享状态项目生命周期管理state_with_team.py当单个步骤承载不了复杂分工时可以把多个 Agent 编成Team作为步骤的 executorTeam 内部成员与 Workflow 共享同一个 session state。state_with_team.py 用项目管理场景演示 Team 步骤与后续 Agent 步骤协同维护steps数组。Team 与 Agent 的工具都直接以RunContext读写状态。Team 侧工具挂在 Team 的tools[add_step, delete_step]上负责增删任务步Agent 侧工具update_step_status、assign_step负责状态流转与改派并支持附带 notesdef update_step_status( run_context: RunContext, step_name: str, new_status: str, notes: str , ) - str: if run_context.session_state is None or steps not in run_context.session_state: return [ERROR] No steps found in workflow session state steps run_context.session_state[steps] for step in steps: if step[name] step_name: old_status step[status] step[status] new_status if notes: step[notes] notes step[last_updated] now return f[OK] Updated step {step_name} status from {old_status} to {new_status} return f[ERROR] Step {step_name} not found in the list关键设计在 Team 的 instructions 里做了职责切割TeamStepManager StepCoordinator只管增删状态更新交给下一阶段的 StatusManager从而避免多 Agent 并发改写同一状态时语义混乱management_team Team( nameManagementTeam, members[step_manager, step_coordinator], tools[add_step, delete_step], instructions[ CRITICAL: Use add_step(step_name, assignee, priority) to add steps., CRITICAL: Use delete_step(step_name) to delete steps., IMPORTANT: You do NOT handle status updates - thats handled by the status manager in the next step., IMPORTANT: Do NOT delete steps when asked to mark them as completed - only delete when explicitly asked to delete., ], )Workflow 将 Team 步骤与 Agent 步骤串联project_workflow Workflow( nameProject Management Workflow, dbSqliteDb(db_filetmp/workflow.db), steps[manage_steps_step, update_status_step], session_state{steps: []}, )Step既可挂agent...也可挂team...二者都会把RunContext.session_state带给其内的每个 executor——这是 Team 成员工具能直接读写 Workflow 状态的链路保证。示例通过五轮对话演示新增任务步 → 改状态并写 notes → 改派并完成 → 追加与联动完成 → 删除与上线辅以print_current_steps辅助函数将get_session_state()渲染成带[PENDING]/[COMPLETED]/[HIGH]等标签的可读清单。会话命名运行后自动生成名称rename_session.py状态与流程之外会话元信息同样可以被工作流维护。rename_session.py 演示了在 run 结束后自动为 session 生成一个可读名称。示例本身是一个研究 → 写作的两步顺序工作流Steps容器article_creation_sequence Steps( namearticle_creation, descriptionComplete article creation workflow from research to writing, steps[research_step, writing_step], ) article_workflow Workflow( descriptionAutomated article creation from research to writing, steps[article_creation_sequence], dbSqliteDb(db_filetmp/workflows.db), ) article_workflow.print_response( inputWrite an article about the benefits of renewable energy, markdownTrue, ) article_workflow.set_session_name(autogenerateTrue) print(fNew session name: {article_workflow.get_session_name()})set_session_name(autogenerateTrue)会根据该 session 的运行内容自动生成名称。在源码libs/agno/agno/workflow/workflow.py中可以看到set_session_name支持三种用法——显式传入session_name、置autogenerateTrue调用生成器、或二者都缺时抛出Exception(Session name is not set)名称最终写入session.session_data[session_name]并save_session持久化读取端由get_session_name()从session_data取回。注意在示例当前代码里Workflow未预先指定session_id需保证运行链路已建立 session源码set_session_name在session_id is None时同样会抛出 Session ID is not set实际集成时可按需显式传入session_id。常用状态 API 速查结合上述示例与 Workflow 源码libs/agno/agno/workflow/workflow.py会话状态相关的常用 API 可归纳如下API / 参数位置与含义Workflow(session_state{...})初始化时的状态种子首次运行 schema存在历史状态时默认加载历史Workflow(overwrite_db_session_stateTrue)显式用本次传入的session_state覆盖库中已存状态Workflow(session_id...)固定会话标识跨多次run/ 跨进程命中同一条会话记录run_context.session_state步骤内读写状态对 Agent/Team 工具即为首个run_context形参workflow.get_session_state()从存储读取该 session 的当前状态字典见 workflow.pyworkflow.update_session_state(updates)以键值字典方式增量更新会话状态见 workflow.pyworkflow.set_session_name(autogenerateTrue)运行后自动/手动命名会话并落库见 workflow.pyworkflow.get_session_name()读取会话名称见 workflow.py在运行语义上有几个值得留意的约定均可在 workflow.py 中找到对应实现佐证状态归属 session 而非 runsession_state存于session.session_data因此同 session 多 run 累积、换 session 隔离是默认行为惰性初始化run_context.session_state可能为None工具函数里统一先做if run_context.session_state is None: run_context.session_state {}再配合setdefault惰性建键是最稳妥的写状态姿势状态与流程双向驱动状态可以被读Condition/Router 的分支依据、summary 统计也可以被写工具调用、executor 副作用由此构造出越用越懂用户的会话级记忆。小结选择哪一种状态接入方式本目录 7 个示例实际上覆盖了状态接入 Workflow 的三种典型形态可按场景取舍Agent/Team 工具读写state_with_agent.py、state_with_team.py、job_application_tracker.py让模型自主决策何时写状态适合交互密集、动作语义清晰的清单/跟踪/CRUD 场景结构化输出 工具 固定session_id的组合能支撑真实的业务累积系统。executor 函数读写state_in_function.py、rename_session.py 的配套能力把确定性逻辑与 LLM 调用封装为可编程步骤适合需要精确计数、拼接 prompt、逐事件流式产出的流程。状态驱动控制流state_in_condition.py、state_in_router.py在 Condition 的evaluator与 Router 的selector中读取状态决定分支目标再在后续步骤回写状态形成自适应、可记忆的用户旅程。无论哪种形态底层都依赖同一事实session_state是挂在 session 上的共享字典随 DBSqliteDb / InMemoryDb 等持久化。理解这一点就可以像本文示例一样用最小的代码把多轮对话记忆、跨步骤协作、自适应路由构建在 agno Workflow 之上。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考