本文将向你展示如何创建一支能够像人类一样解决棘手问题的推理智能体团队。不仅如此，整个智能体系统还连接到 mlflow 提示注册表以提取指令，从而提升智能体本身的可读性。最后，所有输出将使用 qdrant MCP 工具发送到 qdrant，作为长期记忆，以便将来进一步检索。通过本介绍，我们将开始构建整个概念。

架构：

所示系统代表一个强大的Agno推理代理团队（由蓝色框表示），该团队包含一个协调代理和多个推理代理。协调代理充当指挥者，使用协调模式决定使用哪个推理代理，并将任务委托给能够处理问题解决不同方面的专业推理代理。

这些推理代理并非单独工作。它们可以利用自定义工具和内置工具来增强自身能力，如图表底部的黄色框所示。这种工具集成使系统能够超越单纯的自然语言处理，实现更专业的功能。

系统的工作流程从从MLflow提示注册表（顶部的绿色框）中提取指令和提示开始。该注册表可能充当一个集中式的提示策略和模板存储库，内置版本控制功能，用于指导代理的行为和能力。

推理代理完成任务后，最终输出会使用MCP功能推送到Qdrant中，Qdrant充当长期记忆存储和检索系统。

实现：

项目框架如下所示。

.
├── coordinator
│   ├── __init__.py__init__.py
│   └── script.py
├── reasoning_agents
│   ├── __init__.py
│   └── script.py
├── requirements.txt
├── response_agents
│   ├── __init__.py
│   └── script.py
└── utils
    ├── __init__.py
    └── publish2pdf.py

requirements.txt 和 .env 文件的内容大致如下，

#requirements.txt
agno
mlflow
anthropic
python-dotenv
pydantic
yfinance
reportlab
newspaper4k
lxml_html_clean
duckduckgo-search
#.env
OPENAI_API_KEY=sk-proj-
ANTHROPIC_API_KEY=sk-ant-

下面的代码利用 Agno 框架定义了两个专门的 AI 代理：financial_reasoning_agent（财务推理代理）和 company_info_research_reasoning_agent（公司信息研究推理代理）。这些代理分别被设计用于执行财务数据分析和公司信息检索。

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.models.openai import OpenAIChat
from agno.tools.yfinance import YFinanceTools
from agno.tools.thinking import ThinkingTools
from agno.tools.reasoning import ReasoningTools
from pydantic import BaseModel
class CompanyAnalysis(BaseModel):
    company_name: str
    analysis: str
def financial_reasoning_agent(instruction: str) -> Agent:
    _financial_reasoning_agent = Agent(
        name="Stock Searcher",
        role="Searches for information on stocks and provides price analysis.",
        model=Claude(id="claude-3-7-sonnet-20250219"),
        reasoning_model=OpenAIChat(id="o3-mini", reasoning_effort="high"),
        tools=[
            ThinkingTools(),
            ReasoningTools(think=True, add_instructions=True, analyze=True, add_few_shot=True),
            YFinanceTools(
                stock_price=True,
                analyst_recommendations=True,
                company_info=True,
                company_news=True,
            )
        ],
        instructions=instruction,
        use_json_mode=True,
        show_tool_calls=True,
        markdown=True,
    )
    return _financial_reasoning_agent
def company_info_research_reasoning_agent(instruction: str) -> Agent:
    _company_info_agent = Agent(
        name="Company Info Searcher",
        role="Searches for information about companies and recent news.",
        model=Claude(id="claude-3-7-sonnet-20250219"),
        reasoning_model=OpenAIChat(id="o3-mini", reasoning_effort="high"),
        response_model=CompanyAnalysis,
        tools=[
            ThinkingTools(),
            YFinanceTools(
                stock_price=False,
                company_info=True,
                company_news=True,
            )
        ],
        instructions=instruction,
        use_json_mode=True,
        show_tool_calls=True,
        reasoning=True,
        markdown=True,
    )
    return _company_info_agent

该financial_reasoning_agent函数创建了一个名为“Stock Searcher”的代理，其职责是搜索股票信息并提供价格分析。它采用 Anthropic 的 Claude 模型 ( claude-3-7-sonnet-20250219) 执行主要任务，并采用 OpenAI 的 Chat 模型 ( o3-mini) 进行推理，推理工作量设置为较高。该代理配备了 Agno 工具包中的工具：ThinkingTools用于认知处理，ReasoningTools配置为增强分析能力，并可YFinanceTools访问股票价格、分析师建议、公司信息和新闻。该代理配置为以 JSON 模式运行，显示工具调用，并以 Markdown 格式格式化响应。

类似地，该company_info_research_reasoning_agent函数定义了一个名为“公司信息搜索器”的代理，负责检索公司信息和最新新闻。它使用与上一个代理相同的模型，并指定了一个响应模型，它是一个包含 和 字段的PydanticCompanyAnalysis模型。该代理利用了和 一个专注于公司信息和新闻的定制设置。它还配置为以 JSON 模式运行，显示工具调用，启用推理功能，并使用 Markdown 格式格式化响应。BaseModelcompany_nameanalysisThinkingToolsYFinanceTools

以下代码展示了融合协调模式的 AI 代理编排。

路由模式：在路由模式下，协调器充当流量指挥官，分析传入请求并将其发送到具备处理该任务技能的专门代理。想象一个由不同数学运算专家组成的团队——当有计算任务进来时，协调器迅速评估它是加法还是减法问题，并将其转发给相应的数学专家。这种方法通过将问题与最擅长解决它们的代理相匹配，最大限度地提高了效率，而不是让每个代理都尝试执行每个任务，无论其专业水平如何。

协作模式：在这里，在协作模式下，系统同时将相同任务分配给每个代理，允许每个代理使用其独特的能力和专门工具集来解决问题。一旦代理完成其单独工作，协调器就会介入，编译并整合其不同的见解，创建一个统一的解决方案，该方案受益于多种分析方法。这种方法在处理需要不同视角的细微挑战时表现出色，或者当需要同时对同一问题应用不同的分析方法以实现最彻底的理解时。

协调模式：在此配置中，工作流程通过系统按顺序阶段进行，每个代理在前一个代理产生的结果基础上进行构建。协调器监督这一序列，确保从一个代理的输出到下一个代理的输入之间的顺利过渡。考虑一个数据处理场景：第一个代理可能从各种来源收集信息，将其发现传递给第二个代理，该代理对数据进行组织和规范化，然后第三个代理对准备好的数据集执行实际分析。这种方法在任务具有自然依赖关系且必须遵循特定进展顺序以达到预期结果的情况下表现出色。

import asyncio
import mlflow
from agno.models.anthropic import Claude
from agno.team.team import Team
from mcp_agent.script import send_data_to_qdrant
from reasoning_agents.script import financial_reasoning_agent, company_info_research_reasoning_agent
from response_agents.script import writer_agent
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
# Set MLflow tracking URI
mlflow.set_tracking_uri("http://127.0.0.1:3000/")
# fetch the instruction prompt from mlflow prompt registry
# make the prompts more centralized.
fin_thinking_instructions: str = mlflow.load_prompt(name_or_uri='prompts:/thinking_prompt/1').template
fin_coordinator_instructions = mlflow.load_prompt(name_or_uri='prompts:/financial_coordinator_instructions/1').template
writer_agent_instructions = mlflow.load_prompt(name_or_uri='prompts:/writer_agent_instructions/3').template
# initialize the agents by passing the prompt
financial_agent = financial_reasoning_agent(instruction=fin_thinking_instructions)
company_research_agent = company_info_research_reasoning_agent(instruction=fin_thinking_instructions)
article_writer_agent = writer_agent(instructions=writer_agent_instructions)
financial_research_team = Team(
    name="Company & Financial Research Team",
    description="You are a financial expert coordinator that coordinates questions to the appropriate agents. "
                "Finally write a NY Times standards article.",
    mode="coordinate",
    model=Claude("claude-3-7-sonnet-20250219", temperature=0.8),
    members=[financial_agent, company_research_agent, article_writer_agent],
    instructions=fin_coordinator_instructions,
    show_tool_calls=True,
    markdown=True,
    add_datetime_to_instructions=True,
    enable_agentic_context=True,  # Allow the agent to maintain a shared context and send that to members.
    share_member_interactions=True,  # Share all member responses with subsequent member requests.
    show_members_responses=True,
)
async def push_to_memory(context: str) -> None:
    # Create a client session to connect to the MCP server
    agent = await send_data_to_qdrant()
    # Run the agent
    await agent.arun(context)
if __name__ == "__main__":
    response = financial_research_team.run("Write a report comparing NVDA to TSLA")
    asyncio.run(push_to_memory(response.content))
    print(response.content)

环境设置：脚本首先导入必要的库，包括用于实验跟踪的 mlflow、用于创建代理的 Agno 框架组件，以及用于管理环境变量的 dotenv 包中的 load_dotenv。调用 load_dotenv(find_dotenv()) 可确保加载 .env 文件中定义的环境变量，从而便于安全处理敏感信息（如 API 密钥）。

MLflow 配置：脚本将 MLflow 的跟踪 URI 设置为 http://127.0.0.1:3000/，表明 MLflow 的跟踪服务器托管在本地 3000 端口上。此配置允许脚本从 MLflow 服务器记录和检索数据。

从 MLflow 中检索提示：为了集中和管理提示模板，脚本使用 mlflow.load_prompt() 从 MLflow 的提示注册表中获取特定提示。这种方法确保了不同代理在提示使用上的一致性和版本控制。检索到的提示包括：

• fin_thinking_instructions：从名为 'prompts:/thinking_prompt/1' 的提示中加载。
• fin_coordinator_instructions：从 'prompts:/financial_coordinator_instructions/1' 中加载。
• writer_agent_instructions：从 'prompts:/writer_agent_instructions/3' 中加载。

代理初始化：脚本通过传递相应的指令提示来初始化三个代理：

1. 财务代理：使用 financial_reasoning_agent 函数和 fin_thinking_instructions 创建。此代理旨在搜索股票信息并提供价格分析。
2. 公司研究代理：通过 company_info_research_reasoning_agent 函数和 fin_thinking_instructions 初始化。其角色是收集有关公司和最新新闻的信息。
3. 文章撰写代理：使用带有 writer_agent_instructions 的 writer_agent 函数设置。此代理负责根据收集的信息撰写文章。

团队组建：使用 Agno 的 Team 类组建了一个名为“公司与金融研究团队”的团队。该团队充当协调者，将任务分配给适当的代理，并将其输出编译成一份连贯的报告。团队配置包括：

• 模式：设置为“协调”，表明它将在代理之间管理工作流程。
• 模型：使用 Anthropic 的 Claude 模型（claude-3-7-sonnet-20250219），温度设置为 0.8，在响应中平衡创造力和确定性。
• 成员：包括财务代理、公司研究代理和文章撰写代理。
• 指令：由 fin_coordinator_instructions 提供，指导团队的操作。
• 其他设置：配置为显示工具调用、以 Markdown 格式格式化响应、在指令中添加日期时间、启用代理上下文（允许代理之间共享上下文）、共享成员交互并显示成员的响应。

使用 MCP 进行工具交互：

以下代码建立了一个异步函数，用于创建一个与 Qdrant 向量数据库集成的 AI 助手。使用“Agno”框架，它使用特定的配置参数初始化与本地 Qdrant 服务器的连接，然后构建一个由 Claude 3.5 Sonnet 提供支持的代理。该代理被设计为充当“记忆助手”，具有明确的指令，将对话格式化为 JSON，并使用模型上下文协议（MCP）工具将其存储在数据库中。该代码利用环境变量、上下文管理器和专用工具，在 AI 助手的交互与持久存储之间创建无缝桥梁，使系统能够维护对话历史记录或从用户交互中构建知识库。

import asyncio
from textwrap import dedent
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())

async def send_data_to_qdrant():
    # initialize the Qdrant MCP server
    async with MCPTools(
            "uvx mcp-server-qdrant",
            env={
                "QDRANT_URL": "http://localhost:6333",
                "COLLECTION_NAME": "YOUR COLLECTION",
                "QDRANT_API_KEY": "YOUR KEY",
            }
    ) as mcp_tools:
        # Create an agent with the MCP toolkit
        agent = Agent(
            model=Claude(id="claude-3-5-sonnet-20241022"),
            description="memory assistant",
            instructions=dedent("""\
            You are a database expert, your job is to summarize the user query and push the data to 
            database, knowledge base.
            
            1. always create a JSON structure of user query and agent response as 
               example: {'user': <USER_QUERY>, 'assistant': <ASSISTANT_ANSWER>}
            2. after you create the structure always use MCP tools available to push the data into 
               the knowledge base or database.\
        """),
            tools=[mcp_tools],
            markdown=True,
            show_tool_calls=True,
        )
    return agent

当脚本作为主模块执行时，它会提示financial_research_team运行“编写一份比较 NVDA 和 TSLA 的报告”指令。团队的代理协作收集 NVIDIA (NVDA) 和 Tesla (TSLA) 的相关财务数据和公司信息，分析数据并生成一份综合报告。最终输出将打印到控制台，创建 PDF 文件，并最终推送到 Qdrant 以供内存进一步参考。

结论：

本文为在信息技术（IT）领域开发配备推理模型的代理（agents）开辟了新的道路，不仅使代理更具可维护性和可读性，还介绍了将 mlflow prompt registry（MLflow 提示注册表）作为“PromptOps”（提示工程运维）功能进行集成的内容。这是集中式提示管理方面的首创之举。随后，本文还介绍了使用 MCP 集成 Qdrant 以实现记忆（memory）方面的规范化。这些应用可以进行纵向扩展，因为其对于任何用例都更加稳健且易于维护，其中一个用例就是生成 CXO 级别的报告（如示例所示）。