团队是由协作完成任务的智能体(或其他子团队)组成的集合。团队可以通过“协调”、“协作”或“路由”来解决任务。

一个Team有一个members列表,可以是AgentTeam的实例。

from agno.team import Team
from agno.agent import Agent

team = Team(members=[
    Agent(name="Agent 1", role="You answer questions in English"),
    Agent(name="Agent 2", role="You answer questions in Chinese"),
    Team(name="Team 1", role="You answer questions in French"),
])

团队将根据其mode将任务转交给成员。

建议为团队成员指定namerole字段,以便团队领导者更好地识别。

模式

路由模式

路由模式下,团队领导者根据请求内容将用户的请求路由给最合适的团队成员。成员的响应直接返回给用户,团队领导者不解释/转换响应。

async执行中,如果团队领导者一次将任务转交给多个成员,这些成员将并发执行。

协调模式

协调模式下,团队领导者将任务委托给团队成员,并将他们的输出整合为连贯的响应。团队领导者可以根据请求和模型决定最合适的方式,一次性发送给多个成员,或逐个发送。

async执行中,如果团队领导者一次将任务转交给多个成员,这些成员将并发执行。

协作模式

协作模式下,所有团队成员都获得相同的任务,团队领导者将他们的输出整合为连贯的响应。

async执行中,所有成员将并发执行。

团队记忆和历史

团队可以维护先前交互的记忆,从而实现上下文感知:

from agno.team import Team

team_with_memory = Team(
    name="Team with Memory",
    members=[agent1, agent2],
    add_history_to_messages=True,
    num_history_runs=5,
)

# 团队将记住之前的交互
team_with_memory.print_response("What are the key challenges in quantum computing?")
team_with_memory.print_response("Elaborate on the second challenge you mentioned")

团队还可以管理用户记忆:

from agno.team import Team
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory

# 创建一个具有持久化存储的内存实例
memory_db = SqliteMemoryDb(table_name="memory", db_file="memory.db")
memory = Memory(db=memory_db)

team_with_memory = Team(
    name="Team with Memory",
    members=[agent1, agent2],
    memory=memory,
    enable_agentic_memory=True,
)

team_with_memory.print_response("Hi! My name is John Doe.")
team_with_memory.print_response("What is my name?")

团队知识

团队可以使用知识库来存储和检索信息:

from pathlib import Path

from agno.agent import Agent
from agno.embedder.openai import OpenAIEmbedder
from agno.knowledge.url import UrlKnowledge
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.vectordb.lancedb import LanceDb, SearchType

# 设置路径
cwd = Path(__file__).parent
tmp_dir = cwd.joinpath("tmp")
tmp_dir.mkdir(parents=True, exist_ok=True)

# 初始化知识库
agno_docs_knowledge = UrlKnowledge(
    urls=["https://docs.agno.com/llms-full.txt"],
    vector_db=LanceDb(
        uri=str(tmp_dir.joinpath("lancedb")),
        table_name="agno_docs",
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

web_agent = Agent(
    name="Web Search Agent",
    role="Handle web search requests",
    model=OpenAIChat(id="gpt-4o"),
    tools=[DuckDuckGoTools()],
    instructions=["Always include sources"],
)

team_with_knowledge = Team(
    name="Team with Knowledge",
    members=[web_agent],
    model=OpenAIChat(id="gpt-4o"),
    knowledge=agno_docs_knowledge,
    show_members_responses=True,
    markdown=True,
)

if __name__ == "__main__":
    # 在知识库加载后设置为 False
    load_knowledge = True
    if load_knowledge:
        agno_docs_knowledge.load()

    team_with_knowledge.print_response("Tell me about the Agno framework", stream=True)

团队还可以管理用户记忆:

from agno.team import Team
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory

# 创建一个具有持久化存储的内存实例
memory_db = SqliteMemoryDb(table_name="memory", db_file="memory.db")
memory = Memory(db=memory_db)

team_with_memory = Team(
    name="Team with Memory",
    members=[agent1, agent2],
    memory=memory,
    enable_user_memories=True,
)

team_with_memory.print_response("Hi! My name is John Doe.")
team_with_memory.print_response("What is my name?")

会话摘要

要启用会话摘要,请在Team上设置enable_session_summaries=True

from agno.team import Team
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory

team_with_session_summaries = Team(
    name="Team with Memory",
    members=[agent1, agent2],
    enable_session_summaries=True,
)

team_with_session_summaries.print_response("Hi! My name is John Doe and I live in New York City.")

session_summary = team_with_session_summaries.get_session_summary()
print("Session Summary: ", session_summary.summary)

示例

多语言团队

让我们来看一个简单的例子,我们使用不同的模型来回答不同语言的问题。团队由三个专业化的智能体组成,团队领导者将用户的提问路由给合适的语言智能体。

multilanguage_team.py
from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.models.mistral.mistral import MistralChat
from agno.models.openai import OpenAIChat
from agno.team.team import Team

english_agent = Agent(
    name="English Agent",
    role="You only answer in English",
    model=OpenAIChat(id="gpt-4o"),
)
chinese_agent = Agent(
    name="Chinese Agent",
    role="You only answer in Chinese",
    model=DeepSeek(id="deepseek-chat"),
)
french_agent = Agent(
    name="French Agent",
    role="You can only answer in French",
    model=MistralChat(id="mistral-large-latest"),
)

multi_language_team = Team(
    name="Multi Language Team",
    mode="route",
    model=OpenAIChat("gpt-4o"),
    members=[english_agent, chinese_agent, french_agent],
    show_tool_calls=True,
    markdown=True,
    description="You are a language router that directs questions to the appropriate language agent.",
    instructions=[
        "Identify the language of the user's question and direct it to the appropriate language agent.",
        "If the user asks in a language whose agent is not a team member, respond in English with:",
        "'I can only answer in the following languages: English, Chinese, French. Please ask your question in one of these languages.'",
        "Always check the language of the user's input before routing to an agent.",
        "For unsupported languages like Italian, respond in English with the above message.",
    ],
    show_members_responses=True,
)


if __name__ == "__main__":
    # 用所有支持的语言问“你好吗?”
    multi_language_team.print_response("Comment allez-vous?", stream=True)  # 法语
    multi_language_team.print_response("How are you?", stream=True)  # 英语
    multi_language_team.print_response("你好吗?", stream=True)  # 中文
    multi_language_team.print_response("Come stai?", stream=True)  # 意大利语

内容团队

让我们来看另一个例子,我们使用两个专门的智能体来撰写博客文章。团队领导者协调智能体来撰写博客文章。

content_team.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools

# 创建单独的专业智能体
researcher = Agent(
    name="Researcher",
    role="Expert at finding information",
    tools=[DuckDuckGoTools()],
    model=OpenAIChat("gpt-4o"),
)

writer = Agent(
    name="Writer",
    role="Expert at writing clear, engaging content",
    model=OpenAIChat("gpt-4o"),
)

# 创建一个包含这些智能体的团队
content_team = Team(
    name="Content Team",
    mode="coordinate",
    members=[researcher, writer],
    instructions="You are a team of researchers and writers that work together to create high-quality content.",
    model=OpenAIChat("gpt-4o"),
    markdown=True,
)

# 运行团队处理任务
content_team.print_response("Create a short article about quantum computing")

研究团队

这是一个结合了多个专业智能体的研究团队的示例:

1

创建 HackerNews 团队

创建一个文件 hackernews_team.py

hackernews_team.py
from typing import List

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.hackernews import HackerNewsTools
from agno.tools.newspaper4k import Newspaper4kTools
from pydantic import BaseModel

class Article(BaseModel):
    title: str
    summary: str
    reference_links: List[str]


hn_researcher = Agent(
    name="HackerNews Researcher",
    model=OpenAIChat("gpt-4o"),
    role="Gets top stories from hackernews.",
    tools=[HackerNewsTools()],
)

web_searcher = Agent(
    name="Web Searcher",
    model=OpenAIChat("gpt-4o"),
    role="Searches the web for information on a topic",
    tools=[DuckDuckGoTools()],
    add_datetime_to_instructions=True,
)

article_reader = Agent(
    name="Article Reader",
    role="Reads articles from URLs.",
    tools=[Newspaper4kTools()],
)

hackernews_team = Team(
    name="HackerNews Team",
    mode="coordinate",
    model=OpenAIChat("gpt-4o"),
    members=[hn_researcher, web_searcher, article_reader],
    instructions=[
        "First, search hackernews for what the user is asking about.",
        "Then, ask the article reader to read the links for the stories to get more information.",
        "Important: you must provide the article reader with the links to read.",
        "Then, ask the web searcher to search for each story to get more information.",
        "Finally, provide a thoughtful and engaging summary.",
    ],
    response_model=Article,
    show_tool_calls=True,
    markdown=True,
    debug_mode=True,
    show_members_responses=True,
)

# 运行团队
report = hackernews_team.run(
    "What are the top stories on hackernews?"
).content

print(f"Title: {report.title}")
print(f"Summary: {report.summary}")
print(f"Reference Links: {report.reference_links}")
2

运行团队

安装库

pip install openai duckduckgo-search newspaper4k lxml_html_clean agno

运行团队

python hackernews_team.py

开发资源