Agno 智能体会话状态与会话管理实战指南状态、聊天历史与持久化【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno导读本文基于 Agno 仓库cookbook/02_agents/05_state_and_session目录下的 13 个示例系统讲解 Agent 的会话状态session state、聊天历史chat history与会话持久化session persistence三大主题。你将学会如何在工具函数中读写RunContext.session_state、如何让 Agent 感知并自主维护状态、如何把会话写入 SQLite/PostgreSQL 实现跨重启持久化、如何控制上下文窗口大小、如何在多用户场景下隔离会话数据以及如何通过会话摘要压缩上下文。文章同时结合libs/agno/agno/agent/agent.py源码参数默认值与libs/agno/tests/integration/agent/test_session_state.py测试用例为你提供源码级佐证。一、核心概念State、Session、History 三者的关系在 Agno 中一次对话交互称为一个Run运行同一session_id下的多次 Run 构成一个Session会话而 Session 的元数据与消息按user_id/session_id维度存储于数据库。三者的协作关系如下Session State会话状态以字典Dict[str, Any]形式保存的键值数据例如购物清单、用户偏好、计数器等。它是 Agent 的短期记忆/工作记忆默认随每次 Run 落库。Chat History聊天历史会话中产生的用户消息与模型消息序列可注入到上下文让 Agent 记住前面的对话。Persistence持久化将状态与历史写入数据库如SqliteDb、PostgresDb、InMemoryDb、AsyncSqliteDb使得进程重启后会话仍可恢复。围绕这三者示例中出现的核心参数及其源码默认值见 agent.py如下参数默认值作用session_stateNone初始化会话状态的默认字典所有会话的起始状态add_session_state_to_contextFalse是否把会话状态注入模型上下文让 Agent看见当前状态enable_agentic_stateFalse是否允许 Agent 自主管理增删改会话状态search_past_sessionsFalse是否允许 Agent 搜索过往会话num_past_sessions_to_searchNone搜索过往会话时最多纳入的会话数量enable_session_summariesFalse是否启用会话摘要上下文压缩num_history_runsNone实际回退为 3注入上下文的最近 Run 数量store_history_messagesFalse是否把历史消息写入数据库二、环境准备与运行方式所有示例的官方运行前提如下与 README.md 一致加载环境变量运行direnv allow确保包含OPENAI_API_KEY。创建演示环境执行仓库根目录的./scripts/demo_setup.sh之后用.venvs/demo/bin/python运行 cookbook。部分示例需要 PostgreSQL先启动./cookbook/scripts/run_pgvector.sh例如chat_history.py、persistent_session.py、session_summary.py使用postgresqlpsycopg://ai:ailocalhost:5532/ai连接本地 pgvector 容器。统一运行命令模板.venvs/demo/bin/python cookbook/02_agents/05_state_and_session/file.py根据目录内 TEST_LOG.md 的实测记录除last_n_session_messages.py因演示环境缺少aiosqlite依赖而标记 FAIL 外其余示例均测试通过单例耗时约 532 秒。运行异步示例前请自行安装aiosqlite。三、会话状态基础用法在工具中读写状态最直接的状态使用方式是在自定义工具函数中通过RunContext读写session_state。以 session_state_basic.py 为例from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.run import RunContext def add_item(run_context: RunContext, item: str) - str: Add an item to the shopping list. if run_context.session_state is None: run_context.session_state {} run_context.session_state[shopping_list].append(item) return fThe shopping list is now {run_context.session_state[shopping_list]} agent Agent( modelOpenAIResponses(idgpt-5-mini), session_state{shopping_list: []}, # 默认会话状态 dbSqliteDb(db_filetmp/agents.db), # 持久化存储 tools[add_item], instructionsCurrent state (shopping list) is: {shopping_list}, markdownTrue, ) agent.print_response(Add milk, eggs, and bread to the shopping list, streamTrue) print(fFinal session state: {agent.get_session_state()})要点状态注入指令instructions中的{shopping_list}是占位符Agno 会把当前会话状态对应键的值渲染进去让模型知道清单现状。工具通过RunContext改状态工具函数的第一个参数必须是run_context其session_state属性即为当前会话状态字典。由于工具执行发生在 Run 期间改动会在本轮结束后随 Run 一起持久化。读取状态agent.get_session_state()返回合并后的最新状态agent.run()返回的RunOutput也带有session_state字段response.session_state两种读取方式等价。同类参考agent.run()的非流式用法见 session_state_basic.py。四、会话状态进阶构建完整的购物清单管家session_state_advanced.py 把上一节的模式扩展为三个工具构成一个可增、删、查的完整状态机def add_item(run_context: RunContext, item: str) - str: if run_context.session_state is None: run_context.session_state {} if item.lower() not in [i.lower() for i in run_context.session_state[shopping_list]]: run_context.session_state[shopping_list].append(item) return fAdded {item} to the shopping list return f{item} is already in the shopping list def remove_item(run_context: RunContext, item: str) - str: if run_context.session_state is None: run_context.session_state {} for i, list_item in enumerate(run_context.session_state[shopping_list]): if list_item.lower() item.lower(): run_context.session_state[shopping_list].pop(i) return fRemoved {list_item} from the shopping list return f{item} was not found in the shopping list def list_items(run_context: RunContext) - str: if run_context.session_state is None: run_context.session_state {} if not run_context.session_state[shopping_list]: return The shopping list is empty. return Current shopping list:\n \n.join( f- {item} for item in run_context.session_state[shopping_list] )多轮对话验证状态保持agent.print_response(Add milk, eggs, and bread to the shopping list, streamTrue) agent.print_response(I got bread, streamTrue) # 移除 bread agent.print_response(I need apples and oranges, streamTrue) agent.print_response(whats on my list?, streamTrue) agent.print_response( Clear everything from my list and start over with just bananas and yogurt, streamTrue, ) print(fSession state: {agent.get_session_state()})这里体现了两个工程细节幂等性add_item通过小写化比较避免重复添加和友好反馈每个工具返回可读字符串模型可将其转述给用户。这正是把状态管理工具化的推荐姿势——状态只应通过明确定义的函数变更而不是靠模型自由发挥。五、状态合并优先级Run 数据库 Agent 默认值当三个来源的状态同时存在时究竟谁说了算仓库的单元测试 test_session_state.py 给出了明确结论合并优先级session_state_from_run本次 Run 传入session_state_from_db数据库已存self.session_stateAgent 构造默认值具体规则均有测试断言佐证冲突键高优先级覆盖低优先级test_session_state_precedence_all_three_layers非冲突键全部保留DB 独有的db_only、Agent 默认的agent_only都会被合并保留session_stateNone或{}的 Run不会覆盖任何已存状态test_session_state_precedence_empty_run_state_preserves_db嵌套字典同样遵循逐键合并而非整层替换test_session_state_precedence_with_nested_dicts同步、异步arun、流式streamTrue三种运行模式下优先级行为一致test_session_state_precedence_async、test_session_state_precedence_streaming。如果需要整块覆盖而不是合并可在 Agent 上设置overwrite_db_session_stateTrue此时后续 Run 的状态会直接替换掉数据库中已存的旧状态test_session_state_overwriting见 test_session_state.py。六、手动更新会话状态除了工具函数内更新应用代码也可以直接读写状态。核心是两个 API定义于 agent.pyagent.get_session_state(session_idNone)读取可选指定会话的当前状态agent.update_session_state(session_state_updates, session_idNone)写入更新。session_state_manual_update.py 展示了跑完一轮后程序主动追加一项的典型场景agent.print_response(Add milk, eggs, and bread to the shopping list, streamTrue) current_session_state agent.get_session_state() current_session_state[shopping_list].append(chocolate) # 程序侧追加 agent.update_session_state(current_session_state) # 写回并持久化 agent.print_response(Whats on my list?, streamTrue) print(fFinal session state: {agent.get_session_state()})对应测试见 test_session_state.py手动更新后Agent 在下一轮回复中能够正确引用新状态。七、监听状态变更事件当需要把状态变更接入日志、前端推送或审计系统时可以用流式事件机制。关键 API 是RunCompletedEvent配合streamTrue与stream_eventsTruefrom agno.agent import Agent, RunCompletedEvent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses 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 {} run_context.session_state[shopping_list].append(item) return fThe shopping list is now {run_context.session_state[shopping_list]} agent Agent( modelOpenAIResponses(idgpt-5-mini), session_state{shopping_list: []}, dbSqliteDb(db_filetmp/agents.db), tools[add_item], instructionsCurrent state (shopping list) is: {shopping_list}, markdownTrue, ) response agent.run( Add milk, eggs, and bread to the shopping list, streamTrue, stream_eventsTrue, # 开启事件流 ) for event in response: if isinstance(event, RunCompletedEvent): print(fSession state: {event.session_state}) # 事件内携带最终状态见 session_state_events.py。RunCompletedEvent.session_state携带本轮结束后的完整状态快照无需再显式调用get_session_state()。八、Agentic 会话状态让 Agent 自己管理状态前面几节的状态变更都依赖开发者编写工具。Agno 还提供了Agent 自主维护状态的能力只需两个开关enable_agentic_stateTrue允许模型直接增删改会话状态中的字段add_session_state_to_contextTrue必须同时开启否则 Agent 看不到状态也就无从管理。agent Agent( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/agents.db), session_state{shopping_list: []}, add_session_state_to_contextTrue, # 让 Agent 感知状态必需 enable_agentic_stateTrue, # 让 Agent 自主更新状态 ) agent.print_response(Add milk, eggs, and bread to the shopping list) agent.print_response(I picked up the eggs, now whats on my list?) print(fSession state: {agent.get_session_state()})见 agentic_session_state.py。这种模式适合状态结构简单、变更频率低、无需强校验的场景如果状态需要业务约束如去重、权限校验仍建议回归工具函数 RunContext模式。注意源码中两者默认均为Falseagent.py必须显式开启。九、动态会话状态与 Tool Hooks当状态更新逻辑与工具解耦、且需要在工具调用前/后统一拦截时可使用tool_hooks。钩子函数的签名是(run_context: RunContext, arguments: Dict[str, Any])返回字符串作为对工具调用的接管结果此时原工具体不会执行。dynamic_session_state.py 给出了一个客户档案管理示例process_customer_request工具体本身只打印警告并返回 This should not be seen.真正的创建/查询逻辑全部在钩子里完成def customer_management_hook(run_context: RunContext, arguments: Dict[str, Any]): if run_context.session_state is None: run_context.session_state {} action arguments.get(action, retrieve) cust_id arguments.get(customer_id) name arguments.get(name, None) if not cust_id: raise ValueError(customer_id is required.) if action create: run_context.session_state[customer_profiles][cust_id] {name: name} return fSuccess! Customer {cust_id} has been created. if action retrieve: profile run_context.session_state.get(customer_profiles, {}).get(cust_id) if profile: return fProfile for {cust_id}: {json.dumps(profile)} raise ValueError(fCustomer {cust_id} not found.) agent Agent( modelOpenAIResponses(idgpt-5.2), tools[CustomerDBTools()], tool_hooks[customer_management_hook], session_state{customer_profiles: {123: {name: Jane Doe}}}, instructionsYour profiles: {customer_profiles}. Use process_customer_request. Use either create or retrieve as action for the tool., resolve_in_contextTrue, dbInMemoryDb(), )该例还演示了resolve_in_contextTrue工具调用直接在上下文内解析、不做外部执行与InMemoryDb()进程内数据库的组合用法非常适合快速实验钩子逻辑而无需搭建外部存储。十、聊天历史与上下文注入add_history_to_contextTrue会把数据库中的历史消息注入模型上下文使 Agent 拥有多轮记忆。chat_history.py 用 PostgreSQL 存储会话并通过session_idchat_history固定会话db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tablesessions) agent Agent( modelOpenAIResponses(idgpt-5-mini), dbdb, session_idchat_history, instructionsYou are a helpful assistant that can answer questions about space and oceans., add_history_to_contextTrue, # 历史注入上下文 ) agent.print_response(Tell me a new interesting fact about space) print(agent.get_chat_history()) # 读取完整聊天历史 agent.print_response(Tell me a new interesting fact about oceans) print(agent.get_chat_history())get_chat_history(session_idNone, last_n_runsNone)是读取历史的公开 API定义见 agent.py可指定只取最近 N 轮。注意此例固定session_id因此第二次调用仍在同一会话中历史得以延续若每次调用不指定session_idAgno 会生成新的随机会话 ID历史将不共享。十一、会话持久化跨进程重启恢复p persistent_session.py 是本目录最简的持久化示例只要给 Agent 绑定db并固定session_id会话就会自动写入数据库进程重启后用相同session_id再启动 Agent即可恢复此前的对话上下文。agent Agent( modelOpenAIResponses(idgpt-5-mini), dbdb, # PostgresDb绑定存储 session_idsession_storage, # 固定会话 ID实现恢复 add_history_to_contextTrue, ) agent.print_response(Tell me a new interesting fact about space)仓库提供的存储后端覆盖 SQLiteSqliteDb、AsyncSqliteDb、PostgreSQLPostgresDb、内存InMemoryDb以及更多完整示例见 cookbook/06_storage。选择依据单机快速验证用 SQLite生产多实例部署建议 PostgreSQL仅做临时实验可选手内存库。十二、会话选项调优num_history_runs与store_history_messagessession_options.py 演示了用历史但不存历史的精细控制agent Agent( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/example_no_history.db), add_history_to_contextTrue, # 执行时使用历史 num_history_runs3, # 最多注入最近 3 轮 store_history_messagesFalse, # 历史消息不落库 )运行逻辑第一轮让 Agent 记住 My name is Alice and I love Python programming.第二轮问 What is my name and what do I love?Agent 能答出历史参与了推理但agent.get_last_run_output()显示数据库只存了骨架消息历史消息被scrub清洗掉。相关参数默认值agent.pynum_history_runsNone当num_history_messages与num_history_runs都未设置时源码回退默认num_history_runs 3agent.py二者不能同时设置同时设置会告警并优先使用num_history_runsstore_history_messagesFalse默认不把历史消息冗余入库起到省存储、保护隐私的作用。这一节对生产环境意义重大既想保留多轮能力又不想数据库膨胀就可以用读历史 不写历史的组合。十三、限制上下文窗口只注入最近 N 条会话消息长对话的上下文会不断膨胀。目录下另一类限制手段是search_past_sessions体系。 last_n_session_messages.py 用它来限定模型可回顾的会话数量避免上下文长度问题agent Agent( modelOpenAIResponses(idgpt-5-mini), dbAsyncSqliteDb(db_filetmp/data.db), search_past_sessionsTrue, # 允许搜索过往会话 num_past_sessions_to_search2, # 只纳入最近 2 个会话 )脚本构造了两位用户的多个会话User 1 谈首都、User 2 谈人口与货币随后让各自回顾之前聊了什么验证两点一是 Agent 能基于历史会话作答二是搜索按用户隔离——User 1 只会看到自己的会话不会看到 User 2 的内容。异步运行通过asyncio.run(main())与aprint_response(..., session_id..., user_id...)实现。十四、两段式搜索浏览过往会话再精读当会话数量很多时全量注入历史不可行。Agno 提供两步检索模式list-then-read由 search_past_sessions.py 完整演示search_past_sessions()轻量浏览——返回近期每个会话的摘要预览read_past_session(session_id)精读——拉取指定会话的完整对话。启用方式与可选参数源码注释见 search_past_sessions.pyagent Agent( modelOpenAIResponses(idgpt-5.6-luna), dbdb, search_past_sessionsTrue, num_past_sessions_to_search10, # 最多纳入 10 个会话默认 20 # num_past_session_runs_in_search3 # 每个会话在预览中出现的 Run 数默认 3 )示例先写入三个主题会话黑洞、意面、四季再让 Agent 回答我们之前讨论了哪些主题与找到聊烹饪的那个会话Agent 会先search_past_sessions定位再read_past_session读取细节最后用新用户bob提问验证隔离性——bob 没有任何历史Agent 如实回答没有之前对话。该模式适用于知识密集型的长期顾问型 Agent既不撑爆上下文又能按需找回历史。十五、会话摘要把长对话压缩成上下文对超长会话与其注入全部历史不如维护一份会话摘要Session Summary。开启enable_session_summariesTrue后Agno 会在会话推进时持续更新摘要后续 Run 用摘要代替完整历史实现上下文压缩from agno.agent.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIResponses db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tablesessions) # 方法一直接开启摘要 agent Agent( modelOpenAIResponses(idgpt-5-mini), dbdb, enable_session_summariesTrue, session_idsession_123, ) agent.print_response(Hi my name is John and I live in New York) agent.print_response(I like to play basketball and hike in the mountains) print(agent.get_session_summary(session_idsession_123))方法二是注入自定义session_summary_managerSessionSummaryManager(model...)导入自agno.session.summary适合需要指定摘要模型或自定义摘要策略的场景见 session_summary.py。get_session_summary(session_idNone)用于读取当前摘要API 见 agent.py。十六、多用户会话状态隔离生产系统必须隔离不同用户的数据。session_state_multiple_users.py 演示了一个 Agent 实例服务多个用户的隔离模式状态以user_id - session_id - 数据的嵌套字典组织工具函数从run_context.session_state中读取当前用户/会话标识shopping_list {} # user_id - session_id - [items] def add_item(run_context: RunContext, item: str) - str: current_user_id run_context.session_state[current_user_id] current_session_id run_context.session_state[current_session_id] shopping_list.setdefault(current_user_id, {}).setdefault(current_session_id, []).append(item) return fItem {item} added to the shopping list agent Agent( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/data.db), tools[add_item, remove_item, get_shopping_list], instructions[ Current User ID: {current_user_id}, Current Session ID: {current_session_id}, ], markdownTrue, )调用时显式传user_id与session_idagent.print_response(Add milk, eggs, and bread to the shopping list, user_idjohn_doe, session_iduser_1_session_1) agent.print_response(Add tacos to the shopping list, user_idmark_smith, session_iduser_2_session_1)值得注意的两点其一{current_user_id}、{current_session_id}这些占位符由 Agno 在执行时注入到session_state工具据此路由数据其二示例特意演示了新会话即新清单给mark_smith换一个session_id后购物清单从零开始——这正说明user_id负责是谁session_id负责哪次对话两者共同构成数据隔离的边界。十七、结语与进阶路径从工具内读写状态到Agent 自主管理状态从固定会话持久化到搜索历史会话再到摘要压缩上下文本目录的 13 个示例覆盖了会话层几乎所有常见需求。总结选型建议需求推荐参数/能力参考示例状态随对话变化session_state 工具内改RunContext.session_statesession_state_basic.py复杂状态 CRUD多个工具函数 幂等校验session_state_advanced.py程序侧读写状态get_session_state()/update_session_state()session_state_manual_update.py状态变更可观测stream_eventsTrueRunCompletedEventsession_state_events.py模型自主维护状态enable_agentic_stateadd_session_state_to_contextagentic_session_state.py钩子接管状态逻辑tool_hooksresolve_in_contextdynamic_session_state.py多轮记忆add_history_to_contextdbchat_history.py跨重启恢复固定session_iddbpersistent_session.py用历史不存历史store_history_messagesFalsesession_options.py限制回顾范围search_past_sessionsnum_past_sessions_to_searchlast_n_session_messages.py大量历史按需检索search_past_sessions两步模式search_past_sessions.py超长会话压缩enable_session_summaries/SessionSummaryManagersession_summary.py多用户隔离每次运行传user_id/session_idsession_state_multiple_users.py进一步探索状态合并与覆盖的边界行为可阅读测试 test_session_state.py更多存储后端与媒体存储方案见 cookbook/06_storage将这些能力组合到多 Agent 团队或 Workflow 中时可参考 cookbook/03_teams/07_session 与 cookbook/04_workflows。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agno 智能体会话状态与会话管理实战指南：状态、聊天历史与持久化
Agno 智能体会话状态与会话管理实战指南状态、聊天历史与持久化【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno导读本文基于 Agno 仓库cookbook/02_agents/05_state_and_session目录下的 13 个示例系统讲解 Agent 的会话状态session state、聊天历史chat history与会话持久化session persistence三大主题。你将学会如何在工具函数中读写RunContext.session_state、如何让 Agent 感知并自主维护状态、如何把会话写入 SQLite/PostgreSQL 实现跨重启持久化、如何控制上下文窗口大小、如何在多用户场景下隔离会话数据以及如何通过会话摘要压缩上下文。文章同时结合libs/agno/agno/agent/agent.py源码参数默认值与libs/agno/tests/integration/agent/test_session_state.py测试用例为你提供源码级佐证。一、核心概念State、Session、History 三者的关系在 Agno 中一次对话交互称为一个Run运行同一session_id下的多次 Run 构成一个Session会话而 Session 的元数据与消息按user_id/session_id维度存储于数据库。三者的协作关系如下Session State会话状态以字典Dict[str, Any]形式保存的键值数据例如购物清单、用户偏好、计数器等。它是 Agent 的短期记忆/工作记忆默认随每次 Run 落库。Chat History聊天历史会话中产生的用户消息与模型消息序列可注入到上下文让 Agent 记住前面的对话。Persistence持久化将状态与历史写入数据库如SqliteDb、PostgresDb、InMemoryDb、AsyncSqliteDb使得进程重启后会话仍可恢复。围绕这三者示例中出现的核心参数及其源码默认值见 agent.py如下参数默认值作用session_stateNone初始化会话状态的默认字典所有会话的起始状态add_session_state_to_contextFalse是否把会话状态注入模型上下文让 Agent看见当前状态enable_agentic_stateFalse是否允许 Agent 自主管理增删改会话状态search_past_sessionsFalse是否允许 Agent 搜索过往会话num_past_sessions_to_searchNone搜索过往会话时最多纳入的会话数量enable_session_summariesFalse是否启用会话摘要上下文压缩num_history_runsNone实际回退为 3注入上下文的最近 Run 数量store_history_messagesFalse是否把历史消息写入数据库二、环境准备与运行方式所有示例的官方运行前提如下与 README.md 一致加载环境变量运行direnv allow确保包含OPENAI_API_KEY。创建演示环境执行仓库根目录的./scripts/demo_setup.sh之后用.venvs/demo/bin/python运行 cookbook。部分示例需要 PostgreSQL先启动./cookbook/scripts/run_pgvector.sh例如chat_history.py、persistent_session.py、session_summary.py使用postgresqlpsycopg://ai:ailocalhost:5532/ai连接本地 pgvector 容器。统一运行命令模板.venvs/demo/bin/python cookbook/02_agents/05_state_and_session/file.py根据目录内 TEST_LOG.md 的实测记录除last_n_session_messages.py因演示环境缺少aiosqlite依赖而标记 FAIL 外其余示例均测试通过单例耗时约 532 秒。运行异步示例前请自行安装aiosqlite。三、会话状态基础用法在工具中读写状态最直接的状态使用方式是在自定义工具函数中通过RunContext读写session_state。以 session_state_basic.py 为例from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.run import RunContext def add_item(run_context: RunContext, item: str) - str: Add an item to the shopping list. if run_context.session_state is None: run_context.session_state {} run_context.session_state[shopping_list].append(item) return fThe shopping list is now {run_context.session_state[shopping_list]} agent Agent( modelOpenAIResponses(idgpt-5-mini), session_state{shopping_list: []}, # 默认会话状态 dbSqliteDb(db_filetmp/agents.db), # 持久化存储 tools[add_item], instructionsCurrent state (shopping list) is: {shopping_list}, markdownTrue, ) agent.print_response(Add milk, eggs, and bread to the shopping list, streamTrue) print(fFinal session state: {agent.get_session_state()})要点状态注入指令instructions中的{shopping_list}是占位符Agno 会把当前会话状态对应键的值渲染进去让模型知道清单现状。工具通过RunContext改状态工具函数的第一个参数必须是run_context其session_state属性即为当前会话状态字典。由于工具执行发生在 Run 期间改动会在本轮结束后随 Run 一起持久化。读取状态agent.get_session_state()返回合并后的最新状态agent.run()返回的RunOutput也带有session_state字段response.session_state两种读取方式等价。同类参考agent.run()的非流式用法见 session_state_basic.py。四、会话状态进阶构建完整的购物清单管家session_state_advanced.py 把上一节的模式扩展为三个工具构成一个可增、删、查的完整状态机def add_item(run_context: RunContext, item: str) - str: if run_context.session_state is None: run_context.session_state {} if item.lower() not in [i.lower() for i in run_context.session_state[shopping_list]]: run_context.session_state[shopping_list].append(item) return fAdded {item} to the shopping list return f{item} is already in the shopping list def remove_item(run_context: RunContext, item: str) - str: if run_context.session_state is None: run_context.session_state {} for i, list_item in enumerate(run_context.session_state[shopping_list]): if list_item.lower() item.lower(): run_context.session_state[shopping_list].pop(i) return fRemoved {list_item} from the shopping list return f{item} was not found in the shopping list def list_items(run_context: RunContext) - str: if run_context.session_state is None: run_context.session_state {} if not run_context.session_state[shopping_list]: return The shopping list is empty. return Current shopping list:\n \n.join( f- {item} for item in run_context.session_state[shopping_list] )多轮对话验证状态保持agent.print_response(Add milk, eggs, and bread to the shopping list, streamTrue) agent.print_response(I got bread, streamTrue) # 移除 bread agent.print_response(I need apples and oranges, streamTrue) agent.print_response(whats on my list?, streamTrue) agent.print_response( Clear everything from my list and start over with just bananas and yogurt, streamTrue, ) print(fSession state: {agent.get_session_state()})这里体现了两个工程细节幂等性add_item通过小写化比较避免重复添加和友好反馈每个工具返回可读字符串模型可将其转述给用户。这正是把状态管理工具化的推荐姿势——状态只应通过明确定义的函数变更而不是靠模型自由发挥。五、状态合并优先级Run 数据库 Agent 默认值当三个来源的状态同时存在时究竟谁说了算仓库的单元测试 test_session_state.py 给出了明确结论合并优先级session_state_from_run本次 Run 传入session_state_from_db数据库已存self.session_stateAgent 构造默认值具体规则均有测试断言佐证冲突键高优先级覆盖低优先级test_session_state_precedence_all_three_layers非冲突键全部保留DB 独有的db_only、Agent 默认的agent_only都会被合并保留session_stateNone或{}的 Run不会覆盖任何已存状态test_session_state_precedence_empty_run_state_preserves_db嵌套字典同样遵循逐键合并而非整层替换test_session_state_precedence_with_nested_dicts同步、异步arun、流式streamTrue三种运行模式下优先级行为一致test_session_state_precedence_async、test_session_state_precedence_streaming。如果需要整块覆盖而不是合并可在 Agent 上设置overwrite_db_session_stateTrue此时后续 Run 的状态会直接替换掉数据库中已存的旧状态test_session_state_overwriting见 test_session_state.py。六、手动更新会话状态除了工具函数内更新应用代码也可以直接读写状态。核心是两个 API定义于 agent.pyagent.get_session_state(session_idNone)读取可选指定会话的当前状态agent.update_session_state(session_state_updates, session_idNone)写入更新。session_state_manual_update.py 展示了跑完一轮后程序主动追加一项的典型场景agent.print_response(Add milk, eggs, and bread to the shopping list, streamTrue) current_session_state agent.get_session_state() current_session_state[shopping_list].append(chocolate) # 程序侧追加 agent.update_session_state(current_session_state) # 写回并持久化 agent.print_response(Whats on my list?, streamTrue) print(fFinal session state: {agent.get_session_state()})对应测试见 test_session_state.py手动更新后Agent 在下一轮回复中能够正确引用新状态。七、监听状态变更事件当需要把状态变更接入日志、前端推送或审计系统时可以用流式事件机制。关键 API 是RunCompletedEvent配合streamTrue与stream_eventsTruefrom agno.agent import Agent, RunCompletedEvent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses 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 {} run_context.session_state[shopping_list].append(item) return fThe shopping list is now {run_context.session_state[shopping_list]} agent Agent( modelOpenAIResponses(idgpt-5-mini), session_state{shopping_list: []}, dbSqliteDb(db_filetmp/agents.db), tools[add_item], instructionsCurrent state (shopping list) is: {shopping_list}, markdownTrue, ) response agent.run( Add milk, eggs, and bread to the shopping list, streamTrue, stream_eventsTrue, # 开启事件流 ) for event in response: if isinstance(event, RunCompletedEvent): print(fSession state: {event.session_state}) # 事件内携带最终状态见 session_state_events.py。RunCompletedEvent.session_state携带本轮结束后的完整状态快照无需再显式调用get_session_state()。八、Agentic 会话状态让 Agent 自己管理状态前面几节的状态变更都依赖开发者编写工具。Agno 还提供了Agent 自主维护状态的能力只需两个开关enable_agentic_stateTrue允许模型直接增删改会话状态中的字段add_session_state_to_contextTrue必须同时开启否则 Agent 看不到状态也就无从管理。agent Agent( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/agents.db), session_state{shopping_list: []}, add_session_state_to_contextTrue, # 让 Agent 感知状态必需 enable_agentic_stateTrue, # 让 Agent 自主更新状态 ) agent.print_response(Add milk, eggs, and bread to the shopping list) agent.print_response(I picked up the eggs, now whats on my list?) print(fSession state: {agent.get_session_state()})见 agentic_session_state.py。这种模式适合状态结构简单、变更频率低、无需强校验的场景如果状态需要业务约束如去重、权限校验仍建议回归工具函数 RunContext模式。注意源码中两者默认均为Falseagent.py必须显式开启。九、动态会话状态与 Tool Hooks当状态更新逻辑与工具解耦、且需要在工具调用前/后统一拦截时可使用tool_hooks。钩子函数的签名是(run_context: RunContext, arguments: Dict[str, Any])返回字符串作为对工具调用的接管结果此时原工具体不会执行。dynamic_session_state.py 给出了一个客户档案管理示例process_customer_request工具体本身只打印警告并返回 This should not be seen.真正的创建/查询逻辑全部在钩子里完成def customer_management_hook(run_context: RunContext, arguments: Dict[str, Any]): if run_context.session_state is None: run_context.session_state {} action arguments.get(action, retrieve) cust_id arguments.get(customer_id) name arguments.get(name, None) if not cust_id: raise ValueError(customer_id is required.) if action create: run_context.session_state[customer_profiles][cust_id] {name: name} return fSuccess! Customer {cust_id} has been created. if action retrieve: profile run_context.session_state.get(customer_profiles, {}).get(cust_id) if profile: return fProfile for {cust_id}: {json.dumps(profile)} raise ValueError(fCustomer {cust_id} not found.) agent Agent( modelOpenAIResponses(idgpt-5.2), tools[CustomerDBTools()], tool_hooks[customer_management_hook], session_state{customer_profiles: {123: {name: Jane Doe}}}, instructionsYour profiles: {customer_profiles}. Use process_customer_request. Use either create or retrieve as action for the tool., resolve_in_contextTrue, dbInMemoryDb(), )该例还演示了resolve_in_contextTrue工具调用直接在上下文内解析、不做外部执行与InMemoryDb()进程内数据库的组合用法非常适合快速实验钩子逻辑而无需搭建外部存储。十、聊天历史与上下文注入add_history_to_contextTrue会把数据库中的历史消息注入模型上下文使 Agent 拥有多轮记忆。chat_history.py 用 PostgreSQL 存储会话并通过session_idchat_history固定会话db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tablesessions) agent Agent( modelOpenAIResponses(idgpt-5-mini), dbdb, session_idchat_history, instructionsYou are a helpful assistant that can answer questions about space and oceans., add_history_to_contextTrue, # 历史注入上下文 ) agent.print_response(Tell me a new interesting fact about space) print(agent.get_chat_history()) # 读取完整聊天历史 agent.print_response(Tell me a new interesting fact about oceans) print(agent.get_chat_history())get_chat_history(session_idNone, last_n_runsNone)是读取历史的公开 API定义见 agent.py可指定只取最近 N 轮。注意此例固定session_id因此第二次调用仍在同一会话中历史得以延续若每次调用不指定session_idAgno 会生成新的随机会话 ID历史将不共享。十一、会话持久化跨进程重启恢复p persistent_session.py 是本目录最简的持久化示例只要给 Agent 绑定db并固定session_id会话就会自动写入数据库进程重启后用相同session_id再启动 Agent即可恢复此前的对话上下文。agent Agent( modelOpenAIResponses(idgpt-5-mini), dbdb, # PostgresDb绑定存储 session_idsession_storage, # 固定会话 ID实现恢复 add_history_to_contextTrue, ) agent.print_response(Tell me a new interesting fact about space)仓库提供的存储后端覆盖 SQLiteSqliteDb、AsyncSqliteDb、PostgreSQLPostgresDb、内存InMemoryDb以及更多完整示例见 cookbook/06_storage。选择依据单机快速验证用 SQLite生产多实例部署建议 PostgreSQL仅做临时实验可选手内存库。十二、会话选项调优num_history_runs与store_history_messagessession_options.py 演示了用历史但不存历史的精细控制agent Agent( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/example_no_history.db), add_history_to_contextTrue, # 执行时使用历史 num_history_runs3, # 最多注入最近 3 轮 store_history_messagesFalse, # 历史消息不落库 )运行逻辑第一轮让 Agent 记住 My name is Alice and I love Python programming.第二轮问 What is my name and what do I love?Agent 能答出历史参与了推理但agent.get_last_run_output()显示数据库只存了骨架消息历史消息被scrub清洗掉。相关参数默认值agent.pynum_history_runsNone当num_history_messages与num_history_runs都未设置时源码回退默认num_history_runs 3agent.py二者不能同时设置同时设置会告警并优先使用num_history_runsstore_history_messagesFalse默认不把历史消息冗余入库起到省存储、保护隐私的作用。这一节对生产环境意义重大既想保留多轮能力又不想数据库膨胀就可以用读历史 不写历史的组合。十三、限制上下文窗口只注入最近 N 条会话消息长对话的上下文会不断膨胀。目录下另一类限制手段是search_past_sessions体系。 last_n_session_messages.py 用它来限定模型可回顾的会话数量避免上下文长度问题agent Agent( modelOpenAIResponses(idgpt-5-mini), dbAsyncSqliteDb(db_filetmp/data.db), search_past_sessionsTrue, # 允许搜索过往会话 num_past_sessions_to_search2, # 只纳入最近 2 个会话 )脚本构造了两位用户的多个会话User 1 谈首都、User 2 谈人口与货币随后让各自回顾之前聊了什么验证两点一是 Agent 能基于历史会话作答二是搜索按用户隔离——User 1 只会看到自己的会话不会看到 User 2 的内容。异步运行通过asyncio.run(main())与aprint_response(..., session_id..., user_id...)实现。十四、两段式搜索浏览过往会话再精读当会话数量很多时全量注入历史不可行。Agno 提供两步检索模式list-then-read由 search_past_sessions.py 完整演示search_past_sessions()轻量浏览——返回近期每个会话的摘要预览read_past_session(session_id)精读——拉取指定会话的完整对话。启用方式与可选参数源码注释见 search_past_sessions.pyagent Agent( modelOpenAIResponses(idgpt-5.6-luna), dbdb, search_past_sessionsTrue, num_past_sessions_to_search10, # 最多纳入 10 个会话默认 20 # num_past_session_runs_in_search3 # 每个会话在预览中出现的 Run 数默认 3 )示例先写入三个主题会话黑洞、意面、四季再让 Agent 回答我们之前讨论了哪些主题与找到聊烹饪的那个会话Agent 会先search_past_sessions定位再read_past_session读取细节最后用新用户bob提问验证隔离性——bob 没有任何历史Agent 如实回答没有之前对话。该模式适用于知识密集型的长期顾问型 Agent既不撑爆上下文又能按需找回历史。十五、会话摘要把长对话压缩成上下文对超长会话与其注入全部历史不如维护一份会话摘要Session Summary。开启enable_session_summariesTrue后Agno 会在会话推进时持续更新摘要后续 Run 用摘要代替完整历史实现上下文压缩from agno.agent.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIResponses db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tablesessions) # 方法一直接开启摘要 agent Agent( modelOpenAIResponses(idgpt-5-mini), dbdb, enable_session_summariesTrue, session_idsession_123, ) agent.print_response(Hi my name is John and I live in New York) agent.print_response(I like to play basketball and hike in the mountains) print(agent.get_session_summary(session_idsession_123))方法二是注入自定义session_summary_managerSessionSummaryManager(model...)导入自agno.session.summary适合需要指定摘要模型或自定义摘要策略的场景见 session_summary.py。get_session_summary(session_idNone)用于读取当前摘要API 见 agent.py。十六、多用户会话状态隔离生产系统必须隔离不同用户的数据。session_state_multiple_users.py 演示了一个 Agent 实例服务多个用户的隔离模式状态以user_id - session_id - 数据的嵌套字典组织工具函数从run_context.session_state中读取当前用户/会话标识shopping_list {} # user_id - session_id - [items] def add_item(run_context: RunContext, item: str) - str: current_user_id run_context.session_state[current_user_id] current_session_id run_context.session_state[current_session_id] shopping_list.setdefault(current_user_id, {}).setdefault(current_session_id, []).append(item) return fItem {item} added to the shopping list agent Agent( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/data.db), tools[add_item, remove_item, get_shopping_list], instructions[ Current User ID: {current_user_id}, Current Session ID: {current_session_id}, ], markdownTrue, )调用时显式传user_id与session_idagent.print_response(Add milk, eggs, and bread to the shopping list, user_idjohn_doe, session_iduser_1_session_1) agent.print_response(Add tacos to the shopping list, user_idmark_smith, session_iduser_2_session_1)值得注意的两点其一{current_user_id}、{current_session_id}这些占位符由 Agno 在执行时注入到session_state工具据此路由数据其二示例特意演示了新会话即新清单给mark_smith换一个session_id后购物清单从零开始——这正说明user_id负责是谁session_id负责哪次对话两者共同构成数据隔离的边界。十七、结语与进阶路径从工具内读写状态到Agent 自主管理状态从固定会话持久化到搜索历史会话再到摘要压缩上下文本目录的 13 个示例覆盖了会话层几乎所有常见需求。总结选型建议需求推荐参数/能力参考示例状态随对话变化session_state 工具内改RunContext.session_statesession_state_basic.py复杂状态 CRUD多个工具函数 幂等校验session_state_advanced.py程序侧读写状态get_session_state()/update_session_state()session_state_manual_update.py状态变更可观测stream_eventsTrueRunCompletedEventsession_state_events.py模型自主维护状态enable_agentic_stateadd_session_state_to_contextagentic_session_state.py钩子接管状态逻辑tool_hooksresolve_in_contextdynamic_session_state.py多轮记忆add_history_to_contextdbchat_history.py跨重启恢复固定session_iddbpersistent_session.py用历史不存历史store_history_messagesFalsesession_options.py限制回顾范围search_past_sessionsnum_past_sessions_to_searchlast_n_session_messages.py大量历史按需检索search_past_sessions两步模式search_past_sessions.py超长会话压缩enable_session_summaries/SessionSummaryManagersession_summary.py多用户隔离每次运行传user_id/session_idsession_state_multiple_users.py进一步探索状态合并与覆盖的边界行为可阅读测试 test_session_state.py更多存储后端与媒体存储方案见 cookbook/06_storage将这些能力组合到多 Agent 团队或 Workflow 中时可参考 cookbook/03_teams/07_session 与 cookbook/04_workflows。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考