11 - Team 进阶:知识共享、记忆传递与嵌套团队
一句话总结:Team 不只是把 Agent 凑一起,还能共享知识库、传递上下文记忆、甚至把子团队当成员用,搭出真正复杂的协作体系。
上一篇回顾
上一篇我们学了 Team 的基本用法:把几个 Agent 组成团队,用不同的 mode(coordinate、route、broadcast、tasks)来协调分工。那些是”让团队跑起来”的基础。
但在真实项目里,光跑起来远远不够。你会遇到这些问题:
- 团队成员需要访问同一份资料(知识共享)
- 成员之间需要看到彼此的工作结果(上下文传递)
- 团队需要记住之前的对话(会话持久化和记忆)
- 任务太复杂,一个团队搞不定,得拆成多个子团队(嵌套团队)
今天就来逐个解决。
团队共享知识库
知识库(Knowledge)在前面的教程里已经见过了,给单个 Agent 挂一个知识库让它能检索信息。但在团队场景下,一个常见需求是:多个成员共享同一个知识库。
比如一个研究分析团队,研究员和分析师都需要查阅同一批文档。你不需要给每个人建一个独立的知识库,直接共享就行:
from pathlib import Path
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.vectordb.lancedb import LanceDb, SearchType
# 准备目录
tmp_dir = Path("tmp/team_knowledge")
tmp_dir.mkdir(parents=True, exist_ok=True)
# 创建共享知识库
team_knowledge = Knowledge(
vector_db=LanceDb(
uri=str(tmp_dir / "lancedb"),
table_name="team_docs",
search_type=SearchType.hybrid,
embedder=OpenAIEmbedder(id="text-embedding-3-small"),
),
)
# 往知识库里塞点内容(实际项目中你可以 insert 文件、URL 等)
team_knowledge.insert(url="https://docs.agno.com/llms-full.txt")
# 两个成员共享同一个 team_knowledge
researcher = Agent(
name="researcher",
role="搜索和整理信息",
model=OpenAIChat(id="gpt-4o-mini"),
knowledge=team_knowledge,
search_knowledge=True,
instructions="你是研究员,负责从知识库中查找相关信息并整理要点",
)
analyst = Agent(
name="analyst",
role="分析数据给出结论",
model=OpenAIChat(id="gpt-4o-mini"),
knowledge=team_knowledge,
search_knowledge=True,
instructions="你是分析师,基于知识库中的信息进行深度分析",
)
team = Team(
name="research_team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[researcher, analyst],
instructions="协调研究员和分析师完成信息检索和分析任务",
show_members_responses=True,
markdown=True,
)
if __name__ == "__main__":
team.print_response("Agno 框架支持哪些模型提供商?", stream=True)
另一种做法是把知识库挂在 Team 级别,而不是每个成员单独挂。这样 Team Leader 自己就能检索知识,然后把相关信息分发给成员:
team = Team(
name="research_team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[researcher, analyst],
knowledge=team_knowledge, # 知识库挂在团队上
search_knowledge=True,
markdown=True,
)
两种方式各有适用场景:成员需要自己随时查资料,就挂到成员上;Leader 统一分配信息就够了,就挂到团队上。
成员之间的上下文传递
默认情况下,团队成员之间是隔离的。成员 A 做了什么,成员 B 完全不知道。这在简单路由场景下没问题,但在协作场景下就不行了。
Agno 提供了几个开关来控制成员之间的信息流:
share_member_interactions:让成员看到彼此的工作
开启后,当 Leader 依次委派任务给多个成员时,后面的成员能看到前面成员的请求和响应。
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.db.sqlite import SqliteDb
db = SqliteDb(db_file="tmp/support_team.db")
profile_agent = Agent(
name="profile_agent",
role="查询用户资料",
model=OpenAIChat(id="gpt-4o-mini"),
instructions="你负责查询和返回用户的基本资料信息",
)
billing_agent = Agent(
name="billing_agent",
role="处理账单问题",
model=OpenAIChat(id="gpt-4o-mini"),
instructions="你负责回答账单相关的问题,可以参考其他成员提供的用户信息",
)
support_team = Team(
name="support_team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[profile_agent, billing_agent],
db=db,
share_member_interactions=True, # 成员之间可以看到彼此的交互
show_members_responses=True,
markdown=True,
)
if __name__ == "__main__":
support_team.print_response(
"帮我查一下用户张三的资料,然后看看他有没有未付账单",
stream=True,
)
这里的效果是:Leader 先让 profile_agent 去查用户资料,拿到结果后委派 billing_agent 时,billing_agent 能看到 profile_agent 返回的用户信息,不需要重复查询。
add_team_history_to_members:把团队历史传给成员
这个选项把团队级别的对话历史(用户和 Team 之间的交互)传递给成员。适用于多轮对话场景,让成员了解之前聊了什么:
from uuid import uuid4
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.team.mode import TeamMode
from agno.db.sqlite import SqliteDb
db = SqliteDb(db_file="tmp/multilingual_team.db")
cn_agent = Agent(
name="cn_agent",
role="回答中文问题",
model=OpenAIChat(id="gpt-4o-mini"),
)
en_agent = Agent(
name="en_agent",
role="Answer English questions",
model=OpenAIChat(id="gpt-4o-mini"),
)
team = Team(
name="multilingual_team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[cn_agent, en_agent],
mode=TeamMode.route,
db=db,
add_team_history_to_members=True, # 把团队历史传给成员
num_team_history_runs=3, # 传最近 3 轮
markdown=True,
)
if __name__ == "__main__":
session_id = f"chat_{uuid4()}"
# 第一轮用中文
team.print_response("我叫小明,是一名Python开发者", stream=True, session_id=session_id)
# 第二轮切换到英文 -- en_agent 依然知道你叫小明
team.print_response("Tell me a joke using my name", stream=True, session_id=session_id)
关键点:即使第二轮路由到了 en_agent,因为开启了 add_team_history_to_members,en_agent 能看到第一轮中文对话的内容,所以它知道你叫小明。
团队记忆与会话持久化
跟单个 Agent 一样,Team 也支持数据库持久化和记忆管理。
基本的会话持久化
给 Team 接上数据库,加上历史上下文,Team 就能跨多轮对话保持记忆:
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.db.sqlite import SqliteDb
db = SqliteDb(db_file="tmp/team_session.db")
helper = Agent(
name="helper",
model=OpenAIChat(id="gpt-4o-mini"),
)
team = Team(
name="assistant_team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[helper],
db=db,
add_history_to_context=True,
num_history_runs=5,
markdown=True,
)
if __name__ == "__main__":
team.print_response("我在学习 Agno 框架", session_id="learn-001", stream=True)
team.print_response("我在学什么来着?", session_id="learn-001", stream=True)
带 MemoryManager 的长期记忆
如果你想让团队拥有跨 session 的长期记忆,需要接入 MemoryManager:
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.memory import MemoryManager
from agno.db.sqlite import SqliteDb
db = SqliteDb(db_file="tmp/team_memory.db")
memory_manager = MemoryManager(model=OpenAIChat(id="gpt-4o-mini"))
helper = Agent(
name="helper",
model=OpenAIChat(id="gpt-4o-mini"),
)
team = Team(
name="smart_team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[helper],
db=db,
memory_manager=memory_manager,
enable_agentic_memory=True, # 让团队自主决定何时创建/更新记忆
markdown=True,
)
if __name__ == "__main__":
user_id = "xiaoming@example.com"
# 第一个 session:告诉团队一些信息
team.print_response(
"我叫小明,我喜欢周末爬山",
stream=True, user_id=user_id, session_id="session-1",
)
# 第二个 session:团队依然记得你
team.print_response(
"我有什么爱好?",
stream=True, user_id=user_id, session_id="session-2",
)
enable_agentic_memory=True 让团队在每次对话后自动提取关键信息存入长期记忆。下次即使换了一个全新的 session,团队也能”认识”这个用户。
嵌套团队
这是 Team 最强大的特性之一:一个 Team 的成员可以是另一个 Team。Team 本身和 Agent 一样可以作为 members 的元素。
想象一下:你有一个研究团队和一个写作团队,上面还有一个总协调团队来指挥它们。
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
# --- 研究子团队 ---
search_agent = Agent(
name="search_agent",
model=OpenAIChat(id="gpt-4o-mini"),
role="收集原始资料和参考信息",
)
analysis_agent = Agent(
name="analysis_agent",
model=OpenAIChat(id="gpt-4o-mini"),
role="提取关键发现和洞察",
)
research_team = Team(
name="research_team",
members=[search_agent, analysis_agent],
model=OpenAIChat(id="gpt-4o-mini"),
instructions=[
"收集相关信息并总结证据",
"突出关键要点和不确定性",
],
)
# --- 写作子团队 ---
writer_agent = Agent(
name="writer_agent",
model=OpenAIChat(id="gpt-4o-mini"),
role="撰写结构清晰的文稿",
)
editor_agent = Agent(
name="editor_agent",
model=OpenAIChat(id="gpt-4o-mini"),
role="优化文字表达和逻辑结构",
)
writing_team = Team(
name="writing_team",
members=[writer_agent, editor_agent],
model=OpenAIChat(id="gpt-4o-mini"),
instructions=[
"基于研究团队提供的材料撰写和打磨最终输出",
"保持语言简洁,面向决策者",
],
)
# --- 总协调团队 ---
coordinator = Team(
name="coordinator",
members=[research_team, writing_team], # 成员是两个子团队
model=OpenAIChat(id="gpt-4o-mini"),
instructions=[
"协调两个子团队交付完整的成果",
"先让研究团队收集证据,再让写作团队进行综合输出",
],
show_members_responses=True,
markdown=True,
)
if __name__ == "__main__":
coordinator.print_response(
"写一份关于创业团队引入 AI 编程助手的简要分析报告",
stream=True,
)
嵌套团队的执行流程:
- 总协调团队收到任务
- Leader 把”收集资料”的子任务委派给研究子团队
- 研究子团队内部协调 search_agent 和 analysis_agent 完成工作
- 研究结果返回给总协调团队
- Leader 把”撰写报告”的子任务连同研究结果委派给写作子团队
- 写作子团队内部协调 writer_agent 和 editor_agent
- 最终结果汇总返回
用嵌套团队的好处是职责清晰、各团队可以独立调试和复用。不过也别嵌套太深,两到三层就够了,层数越多延迟越高、token 消耗越大。
团队配置速查表
Team 的配置项不少,这里总结一下最常用的几个:
| 配置项 | 作用 | 默认值 |
|---|---|---|
show_members_responses |
在终端显示每个成员的单独响应 | False |
respond_directly |
直接返回成员响应,跳过 Leader 综合 | False |
share_member_interactions |
当前轮次中,后面的成员能看到前面成员的交互 | False |
add_team_history_to_members |
把团队历史对话传给成员 | False |
num_team_history_runs |
传给成员的历史轮数 | 3 |
delegate_to_all_members |
把任务同时发给所有成员 | False |
determine_input_for_members |
Leader 为成员定制输入内容 | True |
场景选型指南
不同的业务场景适合不同的配置组合:
简单路由(客服分流): 用户问题分给最合适的专家,直接返回专家的回答。
team = Team(
mode=TeamMode.route,
respond_directly=True,
)
协作分析(多角度评审): 多个成员各自给出意见,Leader 综合汇总。
team = Team(
mode=TeamMode.coordinate,
show_members_responses=True,
share_member_interactions=True,
)
并行调研(广撒网): 把同一个问题同时扔给所有成员,收集多方结果。
team = Team(
mode=TeamMode.broadcast,
)
流水线处理(前后依赖): 成员的工作有先后顺序,后面的依赖前面的结果。这种场景用 Team 也能做,但如果流程比较固定,用 Workflow 可能更合适。
今天学了什么
回顾一下关键要点:
- 共享知识库:多个成员可以共用同一个 Knowledge 实例,也可以把知识库挂在 Team 级别
- share_member_interactions:让成员看到同一轮次中其他成员的工作结果
- add_team_history_to_members:把团队级别的历史对话传递给成员,解决跨轮次的上下文问题
- 团队记忆:
db+add_history_to_context做会话持久化,MemoryManager+enable_agentic_memory做跨 session 的长期记忆 - 嵌套团队:Team 的 members 可以包含其他 Team,适合职责划分明确的复杂任务
- 别过度设计:能用单个 Agent 解决的问题就别上 Team,能用一层 Team 的就别嵌套
下一篇预告
12 - Workflow:流水线式任务编排
Team 擅长的是”一群人协作完成一个任务”,但有些场景是”一个任务分成明确的几个步骤,按顺序执行”。比如先抓取数据、再清洗、再分析、最后生成报告。这种流水线式的编排,用 Workflow 来做更自然、更可控。下一篇我们就来聊这个。