Agno Team 会话(Session)管理实战:持久化、历史传递、摘要与共享完整指南
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
Agno 的 Team 是多 Agent 协作的核心载体,而 Session(会话)决定了 Team 在多次交互之间能否"记住"自己说过什么、做过什么。本文基于 cookbook/03_teams/07_session 目录下的全部示例,系统讲解 Team 会话的持久化存储、聊天历史注入、会话命名与缓存、自动摘要、历史会话搜索、跨 Agent 共享会话,以及嵌套团队之间的历史传递,并深入 Team 源码 验证每个参数的底层语义。读完本文,你将掌握用 Agno Team 搭建"有记忆、可追溯、可跨会话延续"的多 Agent 应用的完整方案。
前置准备与运行环境
官方示例的 README 明确了三条运行前提,与仓库内其余 cookbook 保持一致:
- 加载环境变量:示例默认使用 OpenAI 模型(
OpenAIResponses),需要先通过direnv allow加载.envrc中的密钥(如OPENAI_API_KEY); - 使用项目虚拟环境:推荐用
.venvs/demo/bin/python来运行 cookbook 示例,例如:.venvs/demo/bin/python cookbook/03_teams/07_session/persistent_session.py - 按需启动外部服务:部分示例依赖外部基础设施——
persistent_session.py、chat_history.py、session_summary.py、session_options.py使用 PostgreSQL(连接串形如postgresql+psycopg://ai:ai@localhost:5532/ai),而search_past_sessions.py、custom_session_summary.py、nested_team_history.py等使用 SQLite 或InMemoryDb。README 明确指出这些额外依赖以文件 docstring 中的说明为准。
会话基础:持久化 Session 与聊天历史
持久化会话:让 Team 跨运行"活下来"
一个 Team 默认是无状态的,但如果给它挂上数据库,会话就能跨进程、跨时间保存。persistent_session.py 演示了最基础的用法:
from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIResponses from agno.team import Team db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" db = PostgresDb(db_url=db_url, session_table="sessions") agent = Agent(model=OpenAIResponses(id="gpt-5-mini")) basic_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[agent], db=db, ) history_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[agent], db=db, add_history_to_context=True, num_history_runs=3, )这里的两个关键参数在 team.py 中有明确定义:
| 参数 | 默认值 | 作用 |
|---|---|---|
db | None | 会话存储后端,BaseDb/AsyncBaseDb类型;不设置则运行结束即丢失 |
add_history_to_context | False | 为True时,把聊天历史中的消息加入发送给模型的消息列表(源码注释:"add_history_to_context=true adds messages from the chat history to the messages list sent to the Model") |
num_history_runs | None | 注入上下文的最近运行(run)次数上限 |
num_history_messages | None | 注入上下文的最近消息条数上限(与num_history_runs二选一或配合使用) |
basic_team与history_team的区别是验证重点:前者虽然也把会话写入了数据库,但没有把历史注入上下文,所以第三轮提问"What have we been talking about?"无法唤起前两轮的记忆;后者通过add_history_to_context=True与num_history_runs=3让 Team 能主动回顾过去三轮对话。这揭示了一个重要设计:持久化与记忆是两个独立的开关——db决定"能否保存",add_history_to_context决定"是否回填"。
读取聊天历史:get_chat_history
chat_history.py 展示了"存取分离"的完整闭环:除了用add_history_to_context把历史喂给模型,还可以通过team.get_chat_history()把历史取出来自己处理:
history_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[agent], db=db, ) 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())同时它还演示了num_history_messages的"限量注入"模式:
limited_history_team = Team( model=OpenAIResponses(id="gpt-5.2"), members=[Agent(model=OpenAIResponses(id="gpt-5.2"))], db=db, add_history_to_context=True, num_history_messages=1, # 只把最近 1 条消息放进上下文 )这样即使会话很长,模型每次也只看到最近 N 条消息,是控制上下文长度、降低 token 开销的实用手段。在源码层面,get_chat_history(session_id, last_n_runs)与异步版本aget_chat_history()均在 team.py 中定义,返回List[Message],last_n_runs可进一步限定只取最近几次运行的对话。
会话选项:命名、内存数据库与会话缓存
session_options.py 集中演示了三个高频会话能力。
会话命名与自动命名
renamable_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[agent], db=postgres_db, ) renamable_team.print_response("Tell me a new interesting fact about space") renamable_team.set_session_name(session_name="Interesting Space Facts") print(renamable_team.get_session_name()) renamable_team.set_session_name(autogenerate=True) # 让模型根据对话内容自动起名 print(renamable_team.get_session_name())set_session_name支持两种用法:显式传入session_name固定名称,或传入autogenerate=True由模型根据会话内容自动生成;随后用get_session_name()读取。对应实现位于 team.py,并同时提供异步版本aset_session_name/aget_session_name。在数据库层面,这是通过 team.py 的会话模块 的generate_session_name调用模型完成的。
内存数据库与会话 ID
in_memory_db = InMemoryDb() in_memory_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[research_agent], db=in_memory_db, add_history_to_context=True, num_history_runs=3, session_id="test_session", # 显式固定会话 ID )InMemoryDb适合原型验证与单进程短生命周期场景——会话仅存在内存中,进程退出即消失。显式指定session_id后,后续调用get_chat_history(session_id="test_session")就能取回同一会话内的消息。示例中用pprint打印[m.model_dump(include={"role", "content"}) for m in chat_history]来观察第一轮、第二轮之后的会话累积情况,是排查会话是否真正落库的便捷手段。
会话缓存
cached_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[research_agent], db=sessions_db, session_id="team_session_cache", add_history_to_context=True, cache_session=True, )cache_session=True会把当前 Team 会话缓存在内存中以加速访问,适用于高频往返同一会话的场景。该参数在 team.py 中的注释为"If True, cache the current Team session in memory for faster access"。
会话摘要:压缩长对话,延续上下文
长会话直接把全部历史注入上下文既贵又容易超限,Agno 提供了"摘要"机制,把过去的对话压缩成结构化摘要。
自动摘要与异步读取
session_summary.py 演示了同步、异步两套摘要流程:
sync_db = PostgresDb(db_url=sync_db_url, session_table="sessions") async_db = AsyncPostgresDb(db_url=async_db_url, session_table="sessions") summary_team = Team( model=OpenAIResponses(id="gpt-5-mini"), members=[sync_agent], db=sync_db, enable_session_summaries=True, # 每轮运行结束后自动生成摘要 ) context_summary_team = Team( model=OpenAIResponses(id="gpt-5-mini"), db=sync_db, session_id="session_summary", add_session_summary_to_context=True, # 把摘要注入上下文 members=[sync_agent], )异步版本通过AsyncPostgresDb(连接串改用postgresql+psycopg_async://)配合aprint_response,并在运行结束后调用aget_session_summary(session_id=...)拉取摘要;返回的SessionSummary对象同时包含summary文本与topics主题列表:
summary = await async_summary_team.aget_session_summary(session_id="async_team_session_summary") if summary: print(f"\nSession Summary: {summary.summary}") if summary.topics: print(f"Topics: {', '.join(summary.topics)}")对应参数在 team.py:enable_session_summaries(每轮运行结束创建/更新会话摘要)、add_session_summary_to_context(是否把摘要注入上下文)、session_summary_manager(自定义摘要管理器)。读取方法为get_session_summary/aget_session_summary(team.py)。
自定义摘要管理器
custom_session_summary.py 进一步演示了用SessionSummaryManager定制摘要行为:
from agno.session import SessionSummaryManager db = SqliteDb( db_file="tmp/team_session_summary.db", session_table="team_summary_sessions", ) summary_manager = SessionSummaryManager(model=OpenAIResponses(id="gpt-5-mini")) sprint_team = Team( name="Sprint Team", model=OpenAIResponses(id="gpt-5-mini"), members=[planner], db=db, session_summary_manager=summary_manager, # 替换默认摘要生成器 add_session_summary_to_context=True, )配合agent.instructions可以引导摘要风格。示例中为规划 Agent 设置了"Build concise, sequenced plan summaries."、"Keep recommendations practical."等指令,从而让"两周期 Sprint 计划"这类业务摘要更贴合领域。运行三轮后通过sprint_team.get_session_summary(session_id="sprint-planning-session")取回摘要,并让 Team 基于摘要回答"下一步最重要动作",展示了"摘要压缩 → 上下文复用 → 延续对话"的完整链路。
搜索历史会话:两步模式(list-then-read)
当团队需要"回想"用户更早的会话(而不只是当前会话)时,search_past_sessions.py 展示了官方推荐的两步模式:先列出预览、再按需读取全文。
db = AsyncSqliteDb(db_file=DB_FILE) team = Team( model=OpenAIResponses(id="gpt-5.6-luna"), members=[], db=db, search_past_sessions=True, num_past_sessions_to_search=10, # 控制搜索最近多少条历史会话 )启用search_past_sessions=True后,Team 会获得两个内置工具(见 team.py 参数注释):
search_past_sessions():返回若干历史会话的轻量逐轮预览;read_past_session(session_id):读取某个具体会话的完整对话。
配套的粒度控制参数(源码注释明确给出默认值):
| 参数 | 默认值 | 作用 |
|---|---|---|
search_past_sessions | False | 是否给 Team 注入历史会话搜索工具 |
num_past_sessions_to_search | None(约 20) | 最多搜索多少条历史会话 |
num_past_session_runs_in_search | None(约 3) | 每条预览里展示几次运行的摘要 |
示例还演示了按用户隔离的会话访问:用户 1 建立user1_session_1..3、用户 2 建立user2_session_1..2,随后通过aprint_response(..., session_id=..., user_id=...)让每个用户只能"浏览并阅读"自己的历史会话,例如"What did I discuss in my previous conversations?"与"Read the full conversation from the session where we discussed China"。这正是多租户场景下"会话归属 + 越权隔离"的参考实现。
与单 Agent 共享会话
share_session_with_agent.py 展示了一个独特的用例:同一个session_id在独立 Agent 与 Team 之间来回流转。
db = InMemoryDb() agent = Agent( name="City Planner Agent", id="city-planner-agent-id", model=OpenAIResponses(id="gpt-5.2"), db=db, # 与 Team 共享同一个数据库 tools=[get_weather, get_activities], add_history_to_context=True, ) team = Team( name="City Planner Team", id="city-planner-team-id", model=OpenAIResponses(id="gpt-5.2"), db=db, # 同一个数据库 members=[weather_agent, activities_agent], add_history_to_context=True, ) session_id = str(uuid.uuid4()) agent.print_response("What is the weather like in Tokyo?", session_id=session_id) team.print_response("What activities can I do there?", session_id=session_id) agent.print_response("What else can you tell me about the city? Should I visit?", session_id=session_id)会话的归属以数据库为锚点:只要 Agent 与 Team 使用同一个db实例,并传入相同的session_id,就能在"单 Agent 模式"与"多 Agent 团队模式"之间无缝切换,交替继续同一段对话。这一模式非常适用于"先由个人助理收集信息,再升级到专家团队深度处理,再回到个人助理汇报"的产品形态。
嵌套团队的历史传递
当 Team 的成员本身是 Team(子团队)时,历史如何传递是关键问题。该目录下有三个层层递进的示例。
子团队自行维护对话历史
nested_team_history.py 展示:父团队把任务委托给嵌套子团队时,子团队通过add_history_to_context=True携带它自己的上一轮对话继续工作。示例用mode="route"的main_team包含 Writer 与 Research Team,三轮追问中 Research Team 必须记得第一轮的研究结论才能回答后续问题。
research_team = Team( name="Research Team", model=OpenAIResponses(id="gpt-5.6-sol"), members=[analyst], add_history_to_context=True, role="Conduct research and analysis", ) main_team = Team( name="Main Team", model=OpenAIResponses(id="gpt-5.6-sol"), members=[writer, research_team], db=db, add_history_to_context=True, mode="route", show_members_responses=True, )add_team_history_to_members:把团队级历史发给成员
nested_team_history_to_members.py 演示的add_team_history_to_members=True语义更精细:父团队委托任务时,会向被委托的成员注入过去运行的文本摘要;对嵌套子团队而言,这份摘要按子团队自己的 id 过滤,因此子团队能回忆起"它自己"处理过的事情,而不是根团队领导的对话。
main_team = Team( name="Main Team", model=OpenAIResponses(id="gpt-5.5"), members=[research_team], db=db, add_team_history_to_members=True, # 把团队历史共享给成员 num_team_history_runs=5, 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_id='research_team'. Never delegate to an individual agent inside a sub-team.", ], )源码注释点明了本意:"This means sending the team-level history to the members, not the agent-level history"(team.py),并配套num_team_history_runs(默认 3,此处调为 5)控制发送的历史运行条数(team.py)。示例特意用指令约束"必须把整个子团队作为单元委托",因为只有走"委托给子团队"这条路,子团队才能拿到自己的历史——如果父团队直接委托到子团队内部的叶子 Agent,就绕过了这条路径。
三层深度嵌套:客服升级场景
nested_team_deep_history.py 把模式推到三层:Support Team(Triage Agent)→ Escalation Team(Technical Support + Expert Team)→ Expert Team(Database Expert + Security Expert)。每层都开启add_history_to_context=True,模拟一次数据库故障的完整工单:客户报障 → 补充"昨天部署新功能后开始超时" → 询问"目前发现了什么、建议如何修复",要求每一层都能回忆起自己的历史处理过程。这是嵌套团队会话在真实客服系统中的一个可直接借鉴的架构蓝本。
会话元数据的三层解析规则
除 README 列出的示例外,同目录的 metadata_resolution.py 补充了会话元数据的合并规则,可视为会话体系的一部分:
team.metadata < session.metadata < call-site metadata即:调用现场(run/arun时传入的metadata)优先级最高,其次为会话中存储的metadata,最低为 Team 构造时的metadata默认值。示例通过预置TeamSession(metadata={"user_tier": "premium", ...})、构造 Team 默认元数据{"user_tier": "free", ...}、再在调用时传{"user_tier": "enterprise"}来验证:同一把键依次被会话覆盖、再被调用现场覆盖;而仅在某一层出现的键(如会话的dark_mode、Team 的team_env)会保留到最终结果中。这为按租户、按请求注入上下文变量提供了确定的优先级语义。
会话 API 速查表
综合 Team 源码 与会话示例,以下是最常用的会话相关参数与方法:
构造参数(Team 实例化)
| 参数 | 默认值 | 说明 |
|---|---|---|
session_id | None(自动生成) | 指定会话 ID,跨调用复用同一会话 |
user_id | None | 会话归属用户,用于多用户隔离 |
db | None | 会话落库后端(Postgres/SQLite/InMemory 等) |
cache_session | False | 内存缓存当前会话以加速访问 |
add_history_to_context | False | 把历史消息注入模型上下文 |
num_history_runs/num_history_messages | None | 控制注入历史的运行数/消息数上限 |
add_team_history_to_members | False | 委托成员时附带团队级历史摘要 |
num_team_history_runs | 3 | 附带给成员的团队历史运行条数 |
search_past_sessions | False | 注入历史会话搜索工具(两步模式) |
num_past_sessions_to_search/num_past_session_runs_in_search | None | 搜索范围与预览条数(默认约 20 / 3) |
enable_session_summaries | False | 每轮结束后自动生成会话摘要 |
session_summary_manager | None | 自定义摘要管理器(SessionSummaryManager) |
add_session_summary_to_context | None | 是否把摘要注入上下文 |
常用方法(同步 / 异步)
| 方法 | 作用 |
|---|---|
set_session_name/aset_session_name | 手动命名或autogenerate=True自动命名会话 |
get_session_name/aget_session_name | 读取会话名称 |
get_chat_history/aget_chat_history | 读取会话的聊天历史(Message列表) |
get_session_summary/aget_session_summary | 读取会话摘要(含summary与topics) |
delete_session/adelete_session | 删除会话(可指定是否删除媒体) |
总结
Agno Team 的会话体系由三个层次构成:存储层(db+session_id/user_id决定会话存哪里、归属谁)、记忆层(add_history_to_context、num_history_runs/messages、add_team_history_to_members决定把多少历史喂给模型,以及enable_session_summaries用摘要替代长历史)、检索层(search_past_sessions两步模式支持跨会话回溯,share_session_with_agent支持 Agent 与 Team 共享同一会话)。嵌套团队场景下,历史按子团队 id 作用域隔离,保证每层都能回忆自己的专属上下文。所有参数与 API 均可在 Team 源码 及其 会话示例目录 中找到完整可运行的对照实现,可直接基于这些示例搭建具备持久记忆与多轮协作能力的生产级多 Agent 系统。
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考