目录

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,
    )

嵌套团队的执行流程:

  1. 总协调团队收到任务
  2. Leader 把”收集资料”的子任务委派给研究子团队
  3. 研究子团队内部协调 search_agent 和 analysis_agent 完成工作
  4. 研究结果返回给总协调团队
  5. Leader 把”撰写报告”的子任务连同研究结果委派给写作子团队
  6. 写作子团队内部协调 writer_agent 和 editor_agent
  7. 最终结果汇总返回

用嵌套团队的好处是职责清晰、各团队可以独立调试和复用。不过也别嵌套太深,两到三层就够了,层数越多延迟越高、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 来做更自然、更可控。下一篇我们就来聊这个。