ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

agno Team 会话(Session)管理实战:聊天历史、持久化、摘要与跨 Agent 共享

agno Team 会话(Session)管理实战:聊天历史、持久化、摘要与跨 Agent 共享 agno Team 会话Session管理实战聊天历史、持久化、摘要与跨 Agent 共享【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本指南以 agno 仓库中cookbook/03_teams/07_session目录下的 Team 会话系列示例及其验证记录TEST_LOG.md为主体系统讲解 agno Team 多 Agent 协作中会话Session的完整生命周期、聊天历史注入与限制、持久化存储、会话命名与缓存、会话摘要、历史会话检索以及会话在 Agent 与 Team 之间的共享等能力。读完本文你将掌握如何在 agno 中为 Team 配置数据库、控制历史上下文长度、生成并复用会话摘要并理解会话 ID、Team ID 与成员历史作用域等底层机制。会话Session在 agno Team 中的地位在 agno 中Team 是协调一组专职 Agent 完成任务的运行单元协调模型coordinator通过delegate_task_to_member工具将子任务委托给成员并把成员输出综合为统一回复。会话Session则是这条多轮交互链路的存档单位——它承载同一用户在多轮运行run之间产生的全部消息使得 Team 在下一轮仍能回忆起上一轮的内容。从验证日志cookbook/03_teams/07_session/TEST_LOG.md中可以看到一次典型会话的生命周期调试输出每一步都有对应的运行时动作DEBUG ****** Team ID: 6108bac2-2f55-4b2f-b7aa-56dc828536ba ******* DEBUG ***** Session ID: 1b5ed4ff-62c7-4a74-ae10-593f959eeb86 ***** DEBUG Creating new TeamSession: 1b5ed4ff-62c7-4a74-ae10-593f959eeb86 DEBUG *** Team Run Start: 0ef1315d-1ca4-4e4a-aa85-5b1f1db3c0de *** DEBUG Processing tools for model DEBUG Added tool delegate_task_to_member DEBUG ------------------ OpenAI Response Start ------------------- ... DEBUG Added RunOutput to Team Session DEBUG Created or updated TeamSession record: 1b5ed4ff-62c7-4a74-ae10-593f959eeb86 DEBUG **** Team Run End: 0ef1315d-1ca4-4e4a-aa85-5b1f1db3c0de ****关键信息包括Team ID标识团队实例Session ID标识会话每次运行有独立的Run ID。首次交互时创建新的TeamSession记录之后每次运行结束都会将RunOutput写入会话并执行Created or updated TeamSession record的持久化动作。团队协调模型会加载坐标模式coordinate mode系统提示词其中明确指示需要成员专长时按成员 ID 委托、委托任务描述需自包含、收到成员回复后综合成单一连贯回答而不是简单拼接。对模型运行本身日志还会输出指标统计METRICS例如Tokens: input442, output587, total1029, reasoning384、Duration: 5.6500s、Tokens per second: 103.8934可用于观察每次运行的 token 成本与吞吐。聊天历史获取与数量限制获取完整聊天历史chat_history.pycookbook/03_teams/07_session/chat_history.py演示了两件事获取 Team 的聊天历史以及限制注入上下文的历史消息数量。from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIResponses from agno.team import Team db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tablesessions) agent Agent(modelOpenAIResponses(idgpt-5-mini)) history_team Team( modelOpenAIResponses(idgpt-5-mini), members[agent], dbdb, ) if __name__ __main__: history_team.print_response(Tell me a new interesting fact about space) print(history_team.get_chat_history()) history_team.print_response(Tell me a new interesting fact about oceans) print(history_team.get_chat_history())运行要点通过get_chat_history()返回当前会话的消息列表从日志可见第二轮运行时出现DEBUG Getting messages from previous runs: 2即从数据库中取回了前一轮的 2 条消息user assistant说明即使未显式开启历史注入会话消息本身也被完整保留在存储中。从日志中序列化的Message对象可以看到完整字段包括role、content、created_at、metricsinput_tokens/output_tokens/total_tokens/reasoning_tokens等、provider_data如 OpenAI 的response_id以及temporary等标记这是后续做会话分析或审计时的数据基础。限制注入上下文的历史数量chat_history.py中的第二个 Team 演示了通过num_history_messages限制注入上下文的消息条数limited_history_team Team( modelOpenAIResponses(idgpt-5.2), members[Agent(modelOpenAIResponses(idgpt-5.2))], dbdb, add_history_to_contextTrue, # 开启历史注入 num_history_messages1, # 只注入最近 1 条消息 )从验证日志可以看到该参数的实际效果第二轮Repeat the last message, but make it much more concise把上一条消息改得更简洁时模型仍能准确引用上一条回答日志中出现Using previous_response_id: resp_0522d87414803e1b0069911ae99e208197aa0dece682442b5e说明 agno 在限制历史条数的同时借助 OpenAI 的previous_response_id机制保留了上下文延续能力。这类限长 引用的组合是控制长会话 token 成本的常用手段。会话选项命名、内存数据库与缓存session_options.pycookbook/03_teams/07_session/session_options.py集中演示了三个会话级选项。1. 手动命名与自动生成会话名renamable_team Team( modelOpenAIResponses(idgpt-5-mini), members[agent], dbpostgres_db, ) renamable_team.print_response(Tell me a new interesting fact about space) renamable_team.set_session_name(session_nameInteresting Space Facts) print(renamable_team.get_session_name()) renamable_team.set_session_name(autogenerateTrue) print(renamable_team.get_session_name())验证日志展示了自动命名的完整过程协调模型收到Please provide a suitable name for this conversation in maximum 5 words的命名任务输入是包含完整系统提示与对话摘要的上下文输出为JWST Early Massive Galaxies随后日志打印DEBUG Generated Session Name: JWST Early Massive Galaxies。手动命名则直接写入指定名称。会话名对用户侧展示历史会话列表非常实用。2. 内存数据库 固定会话 IDfrom agno.db.in_memory import InMemoryDb in_memory_db InMemoryDb() in_memory_team Team( modelOpenAIResponses(idgpt-5-mini), members[research_agent], dbin_memory_db, add_history_to_contextTrue, num_history_runs3, # 注入最近 3 轮运行 session_idtest_session, # 固定会话 ID )InMemoryDb不依赖外部服务适合测试与演示session_idtest_session使同一会话可被多次调用复用日志中可见Creating new TeamSession: test_session与Created or updated TeamSession record: test_session。示例中还演示了用get_chat_history(session_idtest_session)在两次运行后分别打印聊天历史CHAT HISTORY AFTER FIRST RUN/CHAT HISTORY AFTER SECOND RUN验证历史随运行累积第一次运行后历史为 2 条Share a 2 sentence horror story与助手回复第二次运行后历史增长为 4 条且第二轮 What was my first message?能够正确回答Share a 2 sentence horror story证明历史确实参与了上下文。示例代码对获取历史做了 try/except 兜底提示首次运行内存数据库可能没有历史——这是内存态存储的边界行为。3. 会话缓存cache_sessioncached_team Team( modelOpenAIResponses(idgpt-5-mini), members[research_agent], dbsessions_db, session_idteam_session_cache, add_history_to_contextTrue, cache_sessionTrue, )cache_sessionTrue时同一session_id下的历史会被持续累积并注入日志中可见Adding 4 messages from history多轮后历史条数增长以及Using previous_response_id复用机制使连续多轮Tell me a new interesting fact about space的对话能不断给出不同事实FRB 磁陀星、流浪行星、恒星互相偷行星证明长会话下上下文与引用的正确工作。持久化会话把历史真正存下来persistent_session.pycookbook/03_teams/07_session/persistent_session.py演示如何用 PostgreSQL 让 Team 会话跨进程持久化from agno.db.postgres import PostgresDb db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tablesessions) basic_team Team( modelOpenAIResponses(idgpt-5-mini), members[agent], dbdb, ) history_team Team( modelOpenAIResponses(idgpt-5-mini), members[agent], dbdb, add_history_to_contextTrue, num_history_runs3, # 注入最近 3 轮运行 )不开启add_history_to_context时会话消息仍被写入sessions表但不会注入模型上下文开启后每轮都会回读历史。num_history_runs控制回读的运行轮数。示例连续询问三个问题后追问What have we been talking about?验证 Team 能否依据注入的历史概括对话主题。表名通过session_table指定便于与 Agent 会话表隔离管理。仓库中还可以看到更多存储后端cookbook/06_storage/下提供 Postgres、SQLite、MySQL、Mongo、Redis、DynamoDB、Firestore 等说明持久化层是可插拔的。注意本示例依赖本地 PostgreSQL示例默认localhost:5532运行前需保证服务可用。会话摘要为长会话做笔记当会话很长时注入全部历史既贵又容易超限。agno 提供会话摘要能力将历史提炼为结构化摘要供后续上下文复用。自动摘要enable_session_summariessession_summary.pycookbook/03_teams/07_session/session_summary.py同时演示同步与异步两种方式sync_db PostgresDb(db_urlsync_db_url, session_tablesessions) async_db AsyncPostgresDb(db_urlasync_db_url, session_tablesessions) summary_team Team( modelOpenAIResponses(idgpt-5-mini), members[sync_agent], dbsync_db, enable_session_summariesTrue, ) context_summary_team Team( modelOpenAIResponses(idgpt-5-mini), dbsync_db, session_idsession_summary, add_session_summary_to_contextTrue, # 把摘要注入上下文 members[sync_agent], ) async_summary_team Team( modelOpenAIResponses(idgpt-5.2), members[async_agent], dbasync_db, session_idasync_team_session_summary, enable_session_summariesTrue, )enable_session_summariesTrue让 Team 自动为会话生成摘要摘要对象包含summary正文与topics主题列表字段可通过get_session_summary(session_id...)取回。add_session_summary_to_contextTrue则把摘要注入后续运行的上下文让第二轮还能记得第一轮说过什么。异步版本使用AsyncPostgresDb连接串postgresqlpsycopg_async://...与aget_session_summary(session_id...)并通过asyncio.run(run_async_summary_demo())驱动。自定义摘要管理器session_summary_managercustom_session_summary.pycookbook/03_teams/07_session/custom_session_summary.py演示注入自定义的摘要管理器让摘要生成使用独立的模型配置from agno.session import SessionSummaryManager from agno.db.sqlite import SqliteDb db SqliteDb(db_filetmp/team_session_summary.db, session_tableteam_summary_sessions) summary_manager SessionSummaryManager(modelOpenAIResponses(idgpt-5-mini)) sprint_team Team( nameSprint Team, modelOpenAIResponses(idgpt-5-mini), members[planner], dbdb, session_summary_managersummary_manager, add_session_summary_to_contextTrue, )示例围绕双周冲刺规划场景连续提问并在最后调用sprint_team.get_session_summary(session_idsprint-planning-session)打印摘要与主题。这说明摘要管理既可以走默认实现也可以由SessionSummaryManager定制模型与行为。说明在本仓库的验证记录中session_summary.py与custom_session_summary.py因 30 秒超时被标记为 FAIL见下文验证记录解读这属于示例运行环境的超时问题不代表 API 本身不可用——上述用法均与示例源码一致建议在服务与网络稳定的环境中重跑验证。检索历史会话两步走先搜索、再细读search_past_sessions.pycookbook/03_teams/07_session/search_past_sessions.py演示多用户场景下搜索历史会话的官方模式Team 获得两个内置工具search_past_sessions()—— 返回近期会话的轻量预览每个会话若干条 run 摘要read_past_session(session_id)—— 读取指定会话的完整对话。import asyncio import os from agno.db.sqlite import AsyncSqliteDb from agno.models.openai import OpenAIResponses from agno.team import Team DB_FILE tmp/team_session_history.db if os.path.exists(DB_FILE): os.remove(DB_FILE) db AsyncSqliteDb(db_fileDB_FILE) team Team( modelOpenAIResponses(idgpt-5.6-luna), members[], dbdb, search_past_sessionsTrue, num_past_sessions_to_search10, )关键参数在 Team 定义 与 搜索工具实现 中均可核实参数默认值作用search_past_sessionsFalse是否给 Team 挂载search_past_sessions/read_past_session两个工具num_past_sessions_to_search未设置时按 20 处理搜索覆盖的历史会话数量上限num_past_session_runs_in_search3每个会话预览中展示的 run 条数示例脚本的核心场景是用户级隔离user_1创建 3 个会话、user_2创建 2 个会话后分别以各自user_id提问What did I discuss in my previous conversations?验证User 1 只能看到自己的会话、User 2 只能看到自己的会话随后再要求Read the full conversation from the session where we discussed China触发read_past_session的完整读取路径。这个两步模式适合构建让 Agent 记住老用户的记忆型应用。注意本示例使用AsyncSqliteDb需要安装aiosqlite依赖验证记录中search_past_sessions.py正是因缺少aiosqlite模块而报ModuleNotFoundError安装依赖后即可运行。跨 Agent 共享会话同一 session_id 的接力share_session_with_agent.pycookbook/03_teams/07_session/share_session_with_agent.py演示最灵活的用法让单个 Agent 与 Team 共享同一个session_id实现Agent 开场、Team 接力、Agent 收尾的流水线import uuid from agno.agent import Agent from agno.db.in_memory import InMemoryDb from agno.models.openai import OpenAIResponses from agno.team import Team db InMemoryDb() def get_weather(city: str) - str: Get the weather for the given city. return fThe weather in {city} is sunny. def get_activities(city: str) - str: Get the activities for the given city. return fThe activities in {city} are swimming and hiking. agent Agent( nameCity Planner Agent, idcity-planner-agent-id, modelOpenAIResponses(idgpt-5.2), dbdb, tools[get_weather, get_activities], add_history_to_contextTrue, ) weather_agent Agent( nameWeather Agent, idweather-agent-id, modelOpenAIResponses(idgpt-5.2), tools[get_weather], ) activities_agent Agent( nameActivities Agent, idactivities-agent-id, modelOpenAIResponses(idgpt-5.2), tools[get_activities], ) team Team( nameCity Planner Team, idcity-planner-team-id, modelOpenAIResponses(idgpt-5.2), dbdb, members[weather_agent, activities_agent], add_history_to_contextTrue, ) if __name__ __main__: session_id str(uuid.uuid4()) agent.print_response(What is the weather like in Tokyo?, session_idsession_id) team.print_response(What activities can I do there?, session_idsession_id) agent.print_response(What else can you tell me about the city? Should I visit?, session_idsession_id)验证日志清晰呈现了这一接力过程先由city-planner-agent-id单独运行Reading AgentSession/Creating new AgentSession: 9dd8f5d7-...Agent 调用get_weather(cityTokyo)工具得到Tokyo 晴天。Teamcity-planner-team-id使用同一session_id接手日志出现Getting messages from previous runs: 4、Adding 4 messages from historyTeam 的协调模型读到 Agent 之前的气象结论委托成员给出基于晴天的东京活动推荐。再次回到 Agent 时Getting messages from previous runs: 6、Adding 6 messages from historyAgent 能综合天气 活动进一步输出全面的旅行建议街区、美食、交通、潜在缺点、是否值得去。这一模式的核心前提是Agent 与 Team 使用相同的数据库实例和相同的 session_id且都开启add_history_to_contextTrue。它非常适合单 Agent 深度交互 ↔ 多 Agent 并行处理混合的产品形态。嵌套 Team 的历史作用域子团队只记得自己做过的事nested_team_history_to_members.pycookbook/03_teams/07_session/nested_team_history_to_members.py演示嵌套团队父 Team 委托给子 Team场景下的历史作用域语义from agno.db.sqlite import SqliteDb db SqliteDb(db_filetmp/nested_team_history_to_members.db) analyst Agent( nameResearch Analyst, modelOpenAIResponses(idgpt-5.5), roleResearch a topic and report a concise finding, ) research_team Team( nameResearch Team, idresearch_team, # 子团队拥有自己的 ID modelOpenAIResponses(idgpt-5.5), members[analyst], roleHandle all research requests, ) main_team Team( nameMain Team, modelOpenAIResponses(idgpt-5.5), members[research_team], dbdb, add_team_history_to_membersTrue, # 把团队历史分享给被委托的成员 num_team_history_runs5, instructions[ You coordinate sub-teams. Delegate every request to the member with id research_team (the Research Team)., Delegate to the team as a whole using member_idresearch_team. Never delegate to an individual agent inside a sub-team., ], )核心知识点add_team_history_to_membersTrue时父 Team 在委托任务时会注入被委托成员自己的历史摘要。源码中可以看到该逻辑按成员 team_id 过滤历史get_team_history_context(team_idmember_team_id, num_runsteam.num_team_history_runs)见 libs/agno/agno/team/_default_tools.py即每个子团队拿到的历史是以自己 team_id 为作用域的而不是父协调者的历史。验证记录针对 PR #8956 的测试显示两轮对话均委托给member_idresearch_team第一轮子团队研究数字 7第二轮提问Research Team 上次研究了什么数字子团队正确回忆起7及其事实且插桩确认委托路径上触发了get_team_history_context(team_idresearch_team)。num_team_history_runs控制注入子团队的历史运行轮数Team 参数默认值为 3见 libs/agno/agno/team/_init.py。示例特意在指令中要求只委托给整个子团队member_idresearch_team不要委托给子团队内的单个 Agent因为只有委托子团队这一路径才会携带子团队自己的历史。验证记录解读如何读懂这份测试日志TEST_LOG.md本质上是示例脚本的自动化验证记录Validation run 2026-02-15T01:01:12可据此判断各示例的运行状态与排障方向示例文件验证状态说明chat_history.pyPASS聊天历史获取与限长注入正常日志展示完整会话生命周期与 METRICSsession_options.pyPASS会话命名手动/自动、内存 DB、固定 session_id、缓存均正常share_session_with_agent.pyPASSAgent 与 Team 共享 session_id 接力正常nested_team_history_to_members.pyPASS嵌套子团队按自身 team_id 注入历史并正确回忆custom_session_summary.pyFAIL30 秒超时属于运行环境/网络耗时问题persistent_session.pyFAIL30 秒超时属于运行环境/网络耗时问题session_summary.pyFAIL30 秒超时属于运行环境/网络耗时问题search_past_sessions.pyFAILModuleNotFoundError: No module named aiosqlite安装aiosqlite依赖即可文件开头还包含两项全局检查Pattern CheckPASS与OpenAIChat referencesnone用于确保示例遵循 cookbook 的代码规范且不使用已被替代的OpenAIChat旧接口。这一日志本身也是一份可复用的模板验证输出同时保留状态 描述 关键结果含报错堆栈便于回归对比。运行环境与前置条件运行cookbook/03_teams/07_session下示例的通用前置条件来自 README.md通过direnv allow加载环境变量如OPENAI_API_KEY使用项目虚拟环境.venvs/demo/bin/python运行 cookbook 示例部分示例依赖额外服务如 PostgreSQL、SQLite 的aiosqlite、InMemoryDb 则无需外部服务具体以各文件 docstring 说明为准PostgreSQL 示例chat_history.py、persistent_session.py、session_summary.py、session_options.pySQLite 示例custom_session_summary.py、nested_team_history_to_members.py、search_past_sessions.py异步纯内存示例share_session_with_agent.pyInMemoryDb。模型方面示例统一使用OpenAIResponses系列gpt-5-mini、gpt-5.2、gpt-5.5等但 Team 的会话机制与模型无关可替换为仓库中其他受支持模型见cookbook/90_models/下数十种模型集成示例。想要深入源码的读者可重点阅读 Team 参数定义add_history_to_context、num_history_runs、add_team_history_to_members、num_team_history_runs、search_past_sessions、num_past_sessions_to_search、enable_session_summaries、session_summary_manager、add_session_summary_to_context等参数以及 默认工具与历史注入实现search_past_sessions工厂函数、历史/摘要注入逻辑。小结围绕 agno Team 的会话机制本文从验证记录与示例源码两个维度梳理了完整能力集会话生命周期Team ID / Session ID / Run ID 三层标识、聊天历史的读取与限长注入add_history_to_context/num_history_messages/num_history_runs、会话命名与缓存set_session_name/cache_session、PostgreSQL / SQLite / InMemory 三种持久化形态、自动与自定义会话摘要enable_session_summaries/session_summary_manager/add_session_summary_to_context、历史会话的两步检索search_past_sessions/read_past_session、Agent 与 Team 共享session_id的接力模式以及嵌套团队按 team_id 作用域注入历史add_team_history_to_members的精确语义。把握这些机制即可为多 Agent 应用构建能记住、能续聊、能检索、可跨角色共享的会话层。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表